文档章节

Spring整合Rabbitmq

魔法王者安琪拉
 魔法王者安琪拉
发布于 2014/07/09 16:45
字数 1068
阅读 3365
收藏 49
点赞 1
评论 5

没有找到一篇完整的文章介绍Spring如何整合Rabbitmq应用,琢磨一天搞出的一个入门的demo与伙伴们分享.

第一步linux环境下安装rabbitMQ,小弟用的是ubantu,不想浪费太多时间这种安装上

sudo apt-get install rabbitmq-server

默认启动端口5672 测试账户guest 密码guest

官方常用命令,想更改用户的可以参考,这里不作详细说明http://www.rabbitmq.com/man/rabbitmqctl.1.man.html

也可以省去安装115.28.141.137个人私服已配置好

第二步添加依赖包

<dependency>
			<groupId>org.springframework.amqp</groupId>
			<artifactId>spring-rabbit</artifactId>
			<version>1.3.5.RELEASE</version>
		</dependency>
	
		<dependency>
			<groupId>commons-lang</groupId>
			<artifactId>commons-lang</artifactId>
			<version>2.6</version>
		</dependency>

第三步配置生产者工程服务配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:task="http://www.springframework.org/schema/task"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
      http://www.springframework.org/schema/context
      http://www.springframework.org/schema/context/spring-context.xsd
      http://www.springframework.org/schema/integration/ip 
      http://www.springframework.org/schema/integration/ip/spring-integration-ip-2.0.xsd
      http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
	  http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
	  http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
	  http://www.springframework.org/schema/rabbit
      http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
      ">

	<!-- 引入jdbc配置文件 -->
	<context:property-placeholder location="classpath:conf/jdbc.properties" />

	<!-- 扫描文件(自动将servicec层注入) -->
	<context:component-scan base-package="mq.service" />
	<!-- 连接服务配置 -->
	<rabbit:connection-factory id="connectionFactory"
		host="115.28.141.137" username="guest" password="guest" port="5672" />

	<rabbit:admin connection-factory="connectionFactory" />

	<!-- queue 队列声明 -->
	<rabbit:queue id="queue_one" durable="true" auto-delete="false"
		exclusive="false" name="queue_one" />


	<!-- exchange queue binging key 绑定 -->
	<rabbit:direct-exchange name="my-mq-exchange"
		durable="true" auto-delete="false" id="my-mq-exchange">
		<rabbit:bindings>
			<rabbit:binding queue="queue_one" key="queue_one_key" />
		</rabbit:bindings>
	</rabbit:direct-exchange>

	<!-- spring amqp默认的是jackson
	的一个插件,目的将生产者生产的数据转换为json存入消息队列,由于fastjson的速度快于jackson,这里替换为fastjson的一个实现
	-->
	<bean id="jsonMessageConverter" class="mq.util.FastJsonMessageConverter"></bean>

	<!-- spring template声明-->
	<rabbit:template exchange="my-mq-exchange" id="amqpTemplate"
		connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
</beans>

FastJsonMessageConverter 类代码如下

package mq.util;

import java.io.UnsupportedEncodingException;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;

import com.alibaba.druid.support.logging.Log;
import com.alibaba.druid.support.logging.LogFactory;
import com.alibaba.fastjson.JSON;

public class FastJsonMessageConverter  extends AbstractMessageConverter {
    private static Log log = LogFactory.getLog(FastJsonMessageConverter.class);
 
    public static final String DEFAULT_CHARSET = "UTF-8";
 
    private volatile String defaultCharset = DEFAULT_CHARSET;
     
    public FastJsonMessageConverter() {
        super();
        //init();
    }
     
    public void setDefaultCharset(String defaultCharset) {
        this.defaultCharset = (defaultCharset != null) ? defaultCharset
                : DEFAULT_CHARSET;
    }
     
    public Object fromMessage(Message message)
            throws MessageConversionException {
        return null;
    }
     
    @SuppressWarnings("unchecked")
	public <T> T fromMessage(Message message,T t) {
        String json = "";
        try {
            json = new String(message.getBody(),"UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return (T) JSON.parseObject(json, t.getClass());
    }   
     
 
    protected Message createMessage(Object objectToConvert,
            MessageProperties messageProperties)
            throws MessageConversionException {
        byte[] bytes = null;
        try {
            String jsonString = JSON.toJSONString(objectToConvert);
            bytes = jsonString.getBytes(this.defaultCharset);
        } catch (UnsupportedEncodingException e) {
            throw new MessageConversionException(
                    "Failed to convert Message content", e);
        } 
        messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
        messageProperties.setContentEncoding(this.defaultCharset);
        if (bytes != null) {
            messageProperties.setContentLength(bytes.length);
        }
        return new Message(bytes, messageProperties);
 
    }
}

生产者代码Producer

package mq.service;

import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service;   @Service public class ProducerMq {          @Autowired     private AmqpTemplate amqpTemplate;          public void sendDataToCrQueue(Object obj) {         amqpTemplate.convertAndSend("queue_one_key", obj);     }   }

 

第四步配置消费者服务工程配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:task="http://www.springframework.org/schema/task"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
      http://www.springframework.org/schema/context
      http://www.springframework.org/schema/context/spring-context.xsd
      http://www.springframework.org/schema/integration/ip 
      http://www.springframework.org/schema/integration/ip/spring-integration-ip-2.0.xsd
      http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
	  http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
	  http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
	  http://www.springframework.org/schema/rabbit
      http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
      ">

	<!-- 引入jdbc配置文件 -->
	<context:property-placeholder location="classpath:conf/jdbc.properties" />

	<!-- 扫描文件(自动将servicec层注入) -->
	<context:component-scan base-package="mq.service" />
	<task:executor id="taskExecutor" pool-size="1-4" queue-capacity="128" />  
	 <!-- 连接服务配置  -->
   <rabbit:connection-factory id="connectionFactory" host="115.28.141.137" username="guest"
        password="guest" port="5672"  />
         
   <rabbit:admin connection-factory="connectionFactory"/>
    
   <!-- queue 队列声明-->
   <rabbit:queue id="queue_one" durable="true" auto-delete="false" exclusive="false" name="queue_one"/>
    <!-- exchange queue binging key 绑定 -->
    <rabbit:direct-exchange name="my-mq-exchange" durable="true" auto-delete="false" id="my-mq-exchange">
        <rabbit:bindings>
            <rabbit:binding queue="queue_one" key="queue_one_key"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>
    <bean id="queueOneLitener" class="mq.service.QueueOneLitener" />
        <!-- queue litener  观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" task-executor="taskExecutor">
        <rabbit:listener queues="queue_one" ref="queueOneLitener"/>  </rabbit:listener-container>
</beans>

监听器代码queueOneLitener

package mq.service;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
 
public class QueueOneLitener implements  MessageListener{
    public void onMessage(Message message) {
    
        System.out.println(" data :" +new String(message.getBody()));
    }
}

第五步测试

消费者端控制层生产数据

控制层Controller

package mq.controller;

import javax.annotation.Resource;

import mq.service.ProducerMq;

import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

@Controller
public class MessageController {
    @Resource
    private ProducerMq producer;
	@RequestMapping("/producer")
	public String producer() throws Exception {
		for(int i=0;i<100;i++){
			producer.sendDataToCrQueue("data"+i);
		}
		return "index";
	}


}

运行结果

 

 

 


 

 

 

 

© 著作权归作者所有

共有 人打赏支持
魔法王者安琪拉
粉丝 77
博文 90
码字总数 30465
作品 0
深圳
程序员
加载中

评论(5)

小矮子人
小矮子人
online033_login@126.com 求源码 本地启动老是报错
x
xxiee
有源码吗?可以发来学习学习吗?chundingxie@gmail.com
Leoops
Leoops
Good job!
谷粑糖
谷粑糖
z正好这两天在看,谢谢···
花儿笑弯了腰
花儿笑弯了腰
顶顶
阿里云Kubernetes SpringCloud 实践进行时(2): 分布式配置管理

简介 为了更好地支撑日益增长的庞大业务量,我们常常需要把服务进行整合、拆分,使我们的服务不仅能通过集群部署抵挡流量的冲击,又能根据业务在其上进行灵活的扩展。随着分布式的普及、服务...

osswangxining
05/25
0
0
Spring Boot整合RabbitMQ实例

什么是消息? 消息是一个或者多个实体之间沟通的一种方式并且无处不在。 自从计算机发明以来,计算机以多种多样的方式发送消息,消息定义了软硬件或者应用程序之间的沟通方式。消息总是有一个...

英雄有梦没死就别停
06/27
0
0
Spring Boot与RabbitMQ结合实现延迟队列的示例

背景 何为延迟队列? 顾名思义,延迟队列就是进入该队列的消息会被延迟消费的队列。而一般的队列,消息一旦入队了之后就会被消费者马上消费。 场景一:在订单系统中,一个用户下单之后通常有...

xiaomin0322
05/11
0
0
RabbitMQ学习以及与Spring的集成(三)

本文介绍RabbitMQ与Spring的简单集成以及消息的发送和接收。 在RabbitMQ的Spring配置文件中,首先需要增加命名空间。 xmlns:rabbit="http://www.springframework.org/schema/rabbit" 其次是模...

onedotdot
06/18
0
0
Spring Boot+RabbitMQ学习笔记

RabbitMQ RabbitMQ 是一个由 Erlang 语言开发的 的开源实现。最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。 消息队列 先从基础开始,加...

听_风
06/07
0
0
通过实例理解 RabbitMQ 的基本概念

先说下自己开发的实例。 最近在使用 Spring Cloud Config 做分布式配置中心(基于 SVN/Git),当所有服务启动后,SVN/Git 中的配置文件更改后,客户端服务读取的还是旧的配置,并不能实时读取...

xumaojun
05/06
0
0
【Spring Cloud Stream】异步任务

一、前言 前两篇博客提高了用线程池和消息队列才实现异步任务。本篇博客谈一谈用SpringCloud Stream来实现异步任务。 Spring Cloud Stream是一个用来为微服务应用构建消息驱动能力的框架。它...

qq_26545305
05/20
0
0
【Spring Cloud】分布式必学springcloud(十)——消息总线 SpringCloud Bus

一、前言 在上篇博客中,向大家介绍了Spring Cloud Config ,分布式配置中心。它分成了服务端和客户端,服务端提供配置,客户端读取配置,既可以使用svn,也可以使用git存放配置文件,集中管...

kisscatforever
04/25
0
0
【Spring Cloud】分布式必学springcloud(十一)——消息驱动 SpringCloud Stream

一、前言 在前面的博客中,基本上已经把springcloud系列的大部分都介绍过了。如果有太明白的小白,还是建议从小编的第一篇博客进行学习。 在这篇博客中,小白向大家介绍一个消息事件驱动框架...

kisscatforever
04/26
0
0
【消息中间件RabbitMQ】Spring AMQP分析与实战视频教程

视频地址 http://edu.51cto.com/course/10627.html 视频大纲 AMQP协议介绍 RabbitMQ的安装,RabbitMQ web管理,监控管理 Exchange介绍 direct Exchang topic Exchange faont Exchange headers......

刘宗泽
06/26
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

纯Python实现鸢尾属植物数据集神经网络模型[图]

纯Python实现鸢尾属植物数据集神经网络模型[图]: 尝试使用过各大公司推出的植物识别APP吗?比如微软识花、花伴侣等这些APP。当你看到一朵不知道学名的花时,只需要打开植物识别APP,拍摄一张...

原创小博客
3分钟前
0
0
Java 基础 之 常量类型转换

http://www.verejava.com/?id=16992572054911 /*类型转换分为以下几种情况1. 整型类型转换 2. 浮点类型转换3. 布尔类型转换4. 字符类型转换 >>>阅读全文...

全部原谅
3分钟前
0
0
2018安卓巴士开发者大会打造Android技术盛宴

2018安卓巴士开发者大会打造Android技术盛宴2018安卓巴士开发者大会将于8月25日在上海举行,作为中国最具前沿性、专业性的安卓技术会议,将邀请来自爱奇艺、阿里、饿了么等知名企业的一线工程...

逆鳞龙
5分钟前
0
0
Spring框架中的设计模式(二)

Spring框架中的设计模式(二) 原创: 瑞查德-Jack 在 上一篇 中我们在Spring中所谈到的设计模式涉及到了创建模式三剑客和1个行为模式(解释器模式)。这次我们会将眼光更多地关注在具有结构性和...

瑞查德-Jack
5分钟前
0
0
JS基础-DOM Event对象

简介 Event 对象代表事件的状态,比如事件在其中发生的元素、键盘按键的状态、鼠标的位置、鼠标按钮的状态。 事件通常与函数结合使用,函数不会在事件发生前被执行! ==注:详表见《JS基础-...

ZHAO_JH
7分钟前
0
0
tomcat 8.5 远程登录管理页面

1、访问的来源受限注释掉 <?xml version="1.0" encoding="UTF-8"?><!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. S......

xixingzhe
14分钟前
0
0
JSON Web Token - 在Web应用间安全地传递信息

JSON Web Token - 在Web应用间安全地传递信息 Sep 06, 2015 in Engineering JSON Web Token(JWT)是一个非常轻巧的规范。这个规范允许我们使用JWT在用户和服务器之间传递安全可靠的信息。 ...

祖冲之
20分钟前
0
0
Java面试基础篇——第七篇:浅析final关键字

Java有许多关键字,static和final这两个是我们必须掌握的,它们有很多种用法,而且在恰当的环境下使用它们,可以提升程序运行性能,优化程序结构,现在我们先来了解下final关键字。 1. final...

developlee的潇洒人生
27分钟前
1
0
Java安全

安全性问题 DDOS攻击 利用软件开发设计漏洞,进行低成本的攻击消耗服务器CPU,内存资源,例如哈希碰撞攻击,加密,解密,图形处理等计算密集型应用 不能靠用户合理使用解决安全问题 文件上传攻击,...

洋葱94
27分钟前
0
0
IDEA2017环境安装配置

一、下载安装 安装2018版: http://www.jetbrains.com/idea/ 二、获取注册码 1、进到文件夹中:C:\Windows\System32\drivers\etc ,找到hosts文件,用记事本编辑 2、如果没有找到hosts文件,...

一个yuanbeth
29分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部