- 浏览: 100970 次
- 性别:
- 来自: 深圳
文章分类
- 全部博客 (209)
- http://docs.jpush.io/server/java_sdk/ (1)
- SpingMVC ModelAndView (1)
- Model (1)
- Control以及参数传递 (1)
- https://www.alipay.com/ (1)
- 检查指定的字符串列表是否不为空。 (1)
- 转自http://my.oschina.net/rpgmakervx/blog/505434 (1)
- 压缩js (1)
- 含包含 字母数字校验 (1)
- 判断浏览器是ie (1)
- fixbox (0)
- 转自http://www.cnblogs.com/lanzi/archive/2010/10/26/1861338.html (1)
- http://dl2.iteye.com/upload/attachment/0113/2406/afbd8d53-dcad-3afc-8d78-41c1591 (0)
- IDEA (0)
- maven搭建 (0)
- http://www.jeasyuicn.com/api/docTtml/index.htm (1)
- 给tomcat添加运行内存: (1)
- JPUSH推送 (1)
- requestScope (0)
- oracle中plsql安装client10连接数据库 (1)
- 不需要安装oracle11g (1)
- tuikuan (1)
- http://www.hjxmall.com/index.php (1)
- java (1)
- 参考 (1)
- xml签名验签模拟程序 (1)
- 技术网站收集分享 (1)
- Java NIO提供了与标准IO不同的IO工作方式 ------来源于 http://www.iteye.com/magazines/132-Java-NIO (1)
- oracle表查询语句优化 (0)
- oracle (5)
- a (1)
- TenpayHttpClient (2)
- mongodb (1)
- http://www.qcloud.com/product/cmongo.html?utm_source=pcsem1&utm_medium=bdgj46&utm_campaign=baidu (1)
- SpringMVC整合MongoDB开发 (0)
- SpringMVC整合MongoDB开发 https://www.mongodb.org (1)
- Java 语言中常见问题总结 (1)
- 数据库SQL优化大总结 (1)
- 拦截器与过滤器的区别 (1)
- Struts2中拦截器与过滤器的区别及执行顺序 (1)
- Spring声明式事务管理与配置 (0)
- oracle OPERTION (1)
- java 高并发多线程开发 (1)
- Collection 与 map 接口相关集合 (1)
- 多线程开发实践 (1)
- JVM调优总结 转自 http://www.importnew.com/18694.html (1)
- redis 五种数据类型的使用场景 转自于 http://blog.csdn.net/gaogaoshan/article/details/41039581 (1)
- HttpWatch http基础 来自于http://blog.csdn.net/gaogaoshan/article/details/21237555 (1)
- maven 远程仓库 http://blog.csdn.net/gaogaoshan/article/details/40266779 (1)
- 生成Webservice客户端的4中方法 http://blog.csdn.net/gaogaoshan/article/details/8980775 (1)
- http://fgh2011.iteye.com/blog/1564283 (1)
- sleep和wait有什么区别 http://xiaowei2002.iteye.com/blog/2174188 (1)
- JDK中常用包及其类 常见的几种RuntimeException (1)
- Java的运行原理 (1)
- mybatis缓存的使用及理解 http://my.oschina.net/dxqr/blog/123163 (1)
- 软件架构设计分析 (1)
- redis技术总结 (3)
- java面试总结知识点 (1)
- ZooKeeper技术 (1)
- Hadoop (1)
- sso单点登录 (1)
- SpringIOC (1)
- 书签ssssssssssssssssssssss (1)
- spring事务的隔离级别 http://www.cnblogs.com/yangy608/archive/2011/06/29/2093478.html (1)
- 秒杀系统架构分析与实战 http://www.importnew.com/18920.html (1)
- oracle 连接plsql配置 (1)
- maven工程集成springmvc http://blog.csdn.net/fox_lht/article/details/16952683 (1)
- java类序列化与反序列化版本唯一号serialVersionUID (1)
- spring注解用法总结 (1)
- eclipse导入maven项目找不到资源文件方法 (1)
- dubbo (0)
- socket 网络编程 服务器与客户端 编程 (1)
- Thread与Runnable实现线程利用线程插队实现求和操作 (1)
- 生产者与消费者模式 (1)
- JAXB注解 java 关于xml的注解,自动生成xml文件 - @XML*** (1)
- xml 与JAVAbean转换 (1)
- MAP (2)
- decimalToString (1)
- 反编译插件 (0)
- 反编译插件 https://sourceforge.net/projects/jadclipse/?source=typ_redirect (1)
- AWX (1)
- 官网地址Dinp (1)
- public interface ExecutorService extends Executor (1)
- MAIN (1)
- 转自于 http://blog.csdn.net/lufeng20/article/details/24314381 (1)
- JaxbUtil (1)
- HttpXmlClient (0)
- Http post 请求 (1)
- URLDecoder (1)
- spdb (0)
- 订单号生成规则 (1)
- 距离0点毫秒数 (1)
- MyBatis存储过程调用java (1)
- Banks (1)
- 学习网址 (1)
- hots配置oracle数据库配置修改 (1)
- 支付宝文档 (1)
- Jadclipse (1)
- filter (0)
- Filter过滤器 (1)
- 工具类 fixbox (1)
- java quartz (0)
- java quartz 实现定时任务,每个礼拜一到礼拜五定时执行 (1)
- Thread (4)
- getResourceAsStream (1)
- BigData (1)
- 开源架构 (17)
- 架构 (0)
- 文件操作 (2)
- tools (20)
- 测试工具 (1)
- vm (1)
- mq (1)
- blog (1)
- 开源架构搭建技术 (6)
- JAVA细节技术点总结 (1)
- 优化技术 (1)
- LINUX命令使用 (1)
- spring (2)
- 大数据 (1)
- struts2 (1)
- python (1)
- pay (1)
- linux (1)
- 压力测试 (1)
- 代码检查插件 (1)
- 游戏端开发 (1)
- 微信开发相关 (1)
- 搜索引擎 (1)
- 网络技术 (1)
- 分布式部署 (1)
- 堆栈信息 (1)
最新评论
1.1、rabbitMQ的优点(适用范围)
1. 基于erlang语言开发具有高可用高并发的优点,适合集群服务器。
2. 健壮、稳定、易用、跨平台、支持多种语言、文档齐全。
3. 有消息确认机制和持久化机制,可靠性高。
4. 开源
其他MQ的优势:
1. Apache ActiveMQ曝光率最高,但是可能会丢消息。
2. ZeroMQ延迟很低、支持灵活拓扑,但是不支持消息持久化和崩溃恢复。
1.2、几个概念说明
producer&Consumer
producer指的是消息生产者,consumer消息的消费者。
Queue
消息队列,提供了FIFO的处理机制,具有缓存消息的能力。rabbitmq中,队列消息可以设置为持久化,临时或者自动删除。
设置为持久化的队列,queue中的消息会在server本地硬盘存储一份,防止系统crash,数据丢失
设置为临时队列,queue中的数据在系统重启之后就会丢失
设置为自动删除的队列,当不存在用户连接到server,队列中的数据会被自动删除Exchange
Exchange类似于数据通信网络中的交换机,提供消息路由策略。rabbitmq中,producer不是通过信道直接将消息发送给queue,而是先发送给Exchange。一个Exchange可以和多个Queue进行绑定,producer在传递消息的时候,会传递一个ROUTING_KEY,Exchange会根据这个ROUTING_KEY按照特定的路由算法,将消息路由给指定的queue。和Queue一样,Exchange也可设置为持久化,临时或者自动删除。
Exchange有4种类型:direct(默认),fanout, topic, 和headers,不同类型的Exchange转发消息的策略有所区别:
Direct
直接交换器,工作方式类似于单播,Exchange会将消息发送完全匹配ROUTING_KEY的Queue
fanout
广播是式交换器,不管消息的ROUTING_KEY设置为什么,Exchange都会将消息转发给所有绑定的Queue。
topic
主题交换器,工作方式类似于组播,Exchange会将消息转发和ROUTING_KEY匹配模式相同的所有队列,比如,ROUTING_KEY为user.stock的Message会转发给绑定匹配模式为 * .stock,user.stock, * . * 和#.user.stock.#的队列。( * 表是匹配一个任意词组,#表示匹配0个或多个词组)
headers
消息体的header匹配(ignore)
Binding
所谓绑定就是将一个特定的 Exchange 和一个特定的 Queue 绑定起来。Exchange 和Queue的绑定可以是多对多的关系。
virtual host
在rabbitmq server上可以创建多个虚拟的message broker,又叫做virtual hosts (vhosts)。每一个vhost本质上是一个mini-rabbitmq server,分别管理各自的exchange,和bindings。vhost相当于物理的server,可以为不同app提供边界隔离,使得应用安全的运行在不同的vhost实例上,相互之间不会干扰。producer和consumer连接rabbit server需要指定一个vhost。
1.3、消息队列的使用过程
1. 客户端连接到消息队列服务器,打开一个channel。
2. 客户端声明一个exchange,并设置相关属性。
3. 客户端声明一个queue,并设置相关属性。
4. 客户端使用routing key,在exchange和queue之间建立好绑定关系。
5. 客户端投递消息到exchange。
6. exchange接收到消息后,就根据消息的key和已经设置的binding,进行消息路由,将消息投递到一个或多个队列里
rabbitMq.xml配置信息
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 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/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">
<!--配置connection-factory,指定连接rabbit server参数 -->
<rabbit:connection-factory id="connectionFactory"
username="asdf" password="123456" host="10.75.4.25" port="5672" />
<!--定义rabbit template用于数据的接收和发送 -->
<rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
exchange="exchangeTest" />
<!--通过指定下面的admin信息,当前producer中的exchange和queue会在rabbitmq服务器上自动生成 -->
<rabbit:admin connection-factory="connectionFactory" />
<!--定义queue -->
<rabbit:queue name="queueTest" durable="true" auto-delete="false" exclusive="false" />
<!-- 定义direct exchange,绑定queueTest -->
<rabbit:direct-exchange name="exchangeTest" durable="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="queueTest" key="queueTestKey"></rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>
<!-- 消息接收者 -->
<bean id="messageReceiver" class="com.lin.consumer.MessageConsumer"></bean>
<!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener queues="queueTest" ref="messageReceiver"/>
</rabbit:listener-container>
</beans>
消息生产者
package com.lin.producer;
import javax.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;
/**
* 功能概要:消息产生,提交到队列中去
*
* @author linbingwen
* @since 2016年1月15日
*/
@Service
public class MessageProducer {
private Logger logger = LoggerFactory.getLogger(MessageProducer.class);
@Resource
private AmqpTemplate amqpTemplate;
public void sendMessage(Object message){
logger.info("to send message:{}",message);
amqpTemplate.convertAndSend("queueTestKey",message);
}
}
3、消息消费者
package com.lin.consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
/**
* 功能概要:消费接收
*
* @author linbingwen
* @since 2016年1月15日
*/
public class MessageConsumer implements MessageListener {
private Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
@Override
public void onMessage(Message message) {
logger.info("receive message:{}",message);
}
}
1.实现一个消息监听器ReceiveMessageListener.Java
[java] view plain copy
print?在CODE上查看代码片派生到我的代码片
package org.springframework.amqp.core;
/**
* Listener interface to receive asynchronous delivery of Amqp Messages.
*
* @author Mark Pollack
*/
public interface MessageListener {
void onMessage(Message message);
}
2.消费者配置Consumer.xml
[html] view plain copy
print?在CODE上查看代码片派生到我的代码片
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<!-- 连接服务配置 -->
<rabbit:connection-factory id="connectionFactory" host="192.168.36.102" username="admin"
password="admin" port="5672" virtual-host="/" channel-cache-size="5" />
<rabbit:admin connection-factory="connectionFactory"/>
<!-- queue 队列声明-->
<rabbit:queue durable="true" auto-delete="false" exclusive="false" name="spring.queue.tag"/>
<!-- exchange queue binging key 绑定 -->
<rabbit:direct-exchange name="spring.queue.exchange" durable="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="spring.queue.tag" key="spring.queue.tag.key"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<bean id="receiveMessageListener"
class="cn.slimsmart.rabbitmq.demo.spring.tag.ReceiveMessageListener" />
<!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" >
<rabbit:listener queues="spring.queue.tag" ref="receiveMessageListener" />
</rabbit:listener-container>
</beans>
3.生产者配置Producer.xml
[html] view plain copy
print?在CODE上查看代码片派生到我的代码片
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<!-- 连接服务配置 -->
<rabbit:connection-factory id="connectionFactory"
host="192.168.36.102" username="admin" password="admin" port="5672"
virtual-host="/" channel-cache-size="5" />
<rabbit:admin connection-factory="connectionFactory" />
<!-- queue 队列声明 -->
<rabbit:queue durable="true"
auto-delete="false" exclusive="false" name="spring.queue.tag" />
<!-- exchange queue binging key 绑定 -->
<rabbit:direct-exchange name="spring.queue.exchange"
durable="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="spring.queue.tag" key="spring.queue.tag.key" />
</rabbit:bindings>
</rabbit:direct-exchange>
<!-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列,由于Gson的速度快于jackson,这里替换为Gson的一个实现 -->
<bean id="jsonMessageConverter"
class="cn.slimsmart.rabbitmq.demo.spring.tag.Gson2JsonMessageConverter" />
<!-- spring template声明 -->
<rabbit:template id="amqpTemplate" exchange="spring.queue.exchange" routing-key="spring.queue.tag.key"
connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
</beans>
4.消费者启动类ConsumerMain.java
[java] view plain copy
print?在CODE上查看代码片派生到我的代码片
package cn.slimsmart.rabbitmq.demo.spring.tag;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ConsumerMain {
public static void main(String[] args) {
new ClassPathXmlApplicationContext("Consumer.xml");
}
}
5.生产者启动类ProducerMain.java
[java] view plain copy
print?在CODE上查看代码片派生到我的代码片
package cn.slimsmart.rabbitmq.demo.spring.tag;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ProducerMain {
public static void main(String[] args) {
ApplicationContext context = new ClassPathXmlApplicationContext("Producer.xml");
AmqpTemplate amqpTemplate = context.getBean(RabbitTemplate.class);
User user = new User();
user.setName("niuniu");
amqpTemplate.convertAndSend(user);
}
}
1.RabbitMQ简介
RabbitMQ是流行的开源消息队列系统,用erlang语言开发。RabbitMQ是AMQP(高级消息队列协议)的标准实现。
官网:http://www.rabbitmq.com/
2.Spring集成RabbitMQ
2.1 maven配置
//pom.xml
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.5.1</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.4.5.RELEASE</version>
</dependency>
2.2 rabbmitmq配置文件
//rabbitmq-config.properties
mq.host=127.0.0.1
mq.username=test
mq.password=123456
mq.port=5672
mq.vhost=testmq
2.3 Spring配置
//application-mq.xml
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
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/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" >
<description>rabbitmq 连接服务配置</description>
<!-- 连接配置 -->
<rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="${mq.port}" virtual-host="${mq.vhost}"/>
<rabbit:admin connection-factory="connectionFactory"/>
<!-- spring template声明-->
<rabbit:template exchange="amqpExchange" id="amqpTemplate" connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
<!-- 消息对象json转换类 -->
<bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
</beans>
3. 在Spring中使用RabbitMQ
3.1 申明一个消息队列Queue
//application-mq.xml
<rabbit:queue id="test_queue_key" name="test_queue_key" durable="true" auto-delete="false" exclusive="false" />
说明:
durable:是否持久化
exclusive: 仅创建者可以使用的私有队列,断开后自动删除
auto_delete: 当所有消费客户端连接断开后,是否自动删除队列
3.2 交换机定义
//application-mq.xml
<rabbit:direct-exchange name="test-mq-exchange" durable="true" auto-delete="false" id="test-mq-exchange">
<rabbit:bindings>
<rabbit:binding queue="test_queue_key" key="test_queue_key"/>
</rabbit:bindings>
</rabbit:direct-exchange>
说明:
rabbit:direct-exchange:定义exchange模式为direct,意思就是消息与一个特定的路由键完全匹配,才会转发。
rabbit:binding:设置消息queue匹配的key
3.3 发送消息Producer
//MQProducer.java
public interface MQProducer {
/**
* 发送消息到指定队列
* @param queueKey
* @param object
*/
public void sendDataToQueue(String queueKey, Object object);
}
@Service
public class MQProducerImpl implements MQProducer {
@Autowired
private AmqpTemplate amqpTemplate;
private final static Logger LOGGER = Logger.getLogger(MQProducerImpl.class);
/* (non-Javadoc)
* @see com.stnts.tita.rm.api.mq.MQProducer#sendDataToQueue(java.lang.String, java.lang.Object)
*/
@Override
public void sendDataToQueue(String queueKey, Object object) {
try {
amqpTemplate.convertAndSend(queueKey, object);
} catch (Exception e) {
LOGGER.error(e);
}
}
}
说明:
convertAndSend:将Java对象转换为消息发送到匹配Key的交换机中Exchange,由于配置了JSON转换,这里是将Java对象转换成JSON字符串的形式。原文:Convert a Java object to an Amqp Message and send it to a default exchange with a specific routing key.
3.4 异步接收消息Consumer
定义监听器
//QueueListenter.java
@Component
public class QueueListenter implements MessageListener {
@Override
public void onMessage(Message msg) {
try{
System.out.print(msg.toString());
}catch(Exception e){
e.printStackTrace();
}
}
}
监听配置
//application-mq.xml
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto">
<rabbit:listener queues="test_queue" ref="queueListenter"/>
</rabbit:listener-container>
说明:
queues:监听的队列,多个的话用逗号(,)分隔
ref:监听器
3.5 JUnit测试
//TestQueue.java
@RunWith(value = SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {
"classpath:/ApplicationContext/ApplicationContext-mq.xml"})
public class TestQueue{
@Autowired
MQProducer mqProducer;
final String queue_key = "test_queue";
@Test
public void send(){
Map<String,Object> msg = new HashMap()<>;
msg.put("data","hello,rabbmitmq!");
mqProducer.sendDataToQueue(query_key,msg);
}
}
Spring Rabbitmq使用实例
Message Producer
生产者的Rabbitmq.xml配置,这里我们使用rabbit标签来配置rabbitmq客户端。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
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/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">
<!--配置connection-factory,指定连接rabbit server参数-->
<rabbit:connection-factory id="connectionFactory" username="guest" password="guest"
host="localhost"
port="5672"
virtual-host="/"/>
<!--通过指定下面的admin信息,当前producer中的exchange和queue会在rabbitmq服务器上自动生成-->
<rabbit:admin connection-factory="connectionFactory"/>
<!--定义queue-->
<rabbit:queue id="com.mj.test" name="com.mj.test" durable="true" auto-delete="false" exclusive="false"/>
<!-- 定义direct exchange,绑定com.mj.test queue -->
<rabbit:direct-exchange name="myChange" durable="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="com.mj.test" key="hello"></rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>
<!--定义rabbit template用于数据的接收和发送-->
<rabbit:template id="myAmqpTemplate" connection-factory="connectionFactory" exchange="myChange"/>
</beans>
ApplicationContext.xml配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.1.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
http://www.springframework.org/schema/jdbc
http://www.springframework.org/schema/jdbc/spring-jdbc-3.1.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-3.1.xsd">
<!-- 使能AOP-->
<aop:aspectj-autoproxy/>
<!-- 自动装载bean使能-->
<context:component-scan base-package="com.mj.amq"/>
<context:annotation-config/>
<!-- 声明一个事物管理 -->
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"></property>
</bean>
<!-- 启用annotation的事务支持 -->
<tx:annotation-driven transaction-manager="transactionManager" proxy-target-class="true"/>
<import resource="Rabbitmq.xml"/>
<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://localhost:3307/student?useUnicode=true&characoding=UTF-8"/>
<property name="username" value="root"/>
<property name="password" value="root"/>
</bean>
</beans>
Producer发送数据Java代码
@Service
public class MessageSender {
@Resource(name="myAmqpTemplate")
AmqpTemplate amqpTemplate;
public void sendMessage(Object message){
amqpTemplate.convertAndSend("hello",message);
}
}
Message Consumer
Rabbitmq.xml配置
Rabbitmq接收端需要配置Connection-Factory实例,监听的队列,以及Listener-container。同时,我们需要创建监听队列的java bean。这里我们使用的是spring rabbitmq提供的异步MessageListener接口,consumer的业务逻辑在onMessage中实现。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
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/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">
<rabbit:connection-factory id="connectionFactory" username="guest" password="guest"
host="localhost"
port="5672"
virtual-host="/"/>
<rabbit:queue id="com.mj.test" name="com.mj.test" durable="true" auto-delete="false" exclusive="false"/>
<bean id="messageReceiver" class="com.mj.amq.MessageReceiver"></bean>
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener queues="com.mj.test" ref="messageReceiver"/>
</rabbit:listener-container>
</beans>
ApplicationContext.xml和producer的一致
Consumer侦听mq消息代码
public class MessageReceiver implements MessageListener {
public void onMessage(Message message) {
System.out.println(message);
}
public static void main(String[]args){
ApplicationContext applicationContext=new ClassPathXmlApplicationContext("ApplicationContext.xml");
}
}
测试
启动MessageReceiver的main函数,运行producer的单元测试,会在MessageReceiver的concole端看到接收都到的数据。同时登陆rabbitmq客户端可以看到我们配置的Exchange和queue信息。
public class MessageSenderTest {
private ApplicationContext context = null;
@Before
public void setUp() throws Exception {
context = new ClassPathXmlApplicationContext("ApplicationContext.xml");
}
@Test
public void should_send_a_amq_message() throws Exception {
MessageSender messageSender = (MessageSender) context.getBean("messageSender");
messageSender.sendMessage("Hello, I am amq sender");
}
}
1. 基于erlang语言开发具有高可用高并发的优点,适合集群服务器。
2. 健壮、稳定、易用、跨平台、支持多种语言、文档齐全。
3. 有消息确认机制和持久化机制,可靠性高。
4. 开源
其他MQ的优势:
1. Apache ActiveMQ曝光率最高,但是可能会丢消息。
2. ZeroMQ延迟很低、支持灵活拓扑,但是不支持消息持久化和崩溃恢复。
1.2、几个概念说明
producer&Consumer
producer指的是消息生产者,consumer消息的消费者。
Queue
消息队列,提供了FIFO的处理机制,具有缓存消息的能力。rabbitmq中,队列消息可以设置为持久化,临时或者自动删除。
设置为持久化的队列,queue中的消息会在server本地硬盘存储一份,防止系统crash,数据丢失
设置为临时队列,queue中的数据在系统重启之后就会丢失
设置为自动删除的队列,当不存在用户连接到server,队列中的数据会被自动删除Exchange
Exchange类似于数据通信网络中的交换机,提供消息路由策略。rabbitmq中,producer不是通过信道直接将消息发送给queue,而是先发送给Exchange。一个Exchange可以和多个Queue进行绑定,producer在传递消息的时候,会传递一个ROUTING_KEY,Exchange会根据这个ROUTING_KEY按照特定的路由算法,将消息路由给指定的queue。和Queue一样,Exchange也可设置为持久化,临时或者自动删除。
Exchange有4种类型:direct(默认),fanout, topic, 和headers,不同类型的Exchange转发消息的策略有所区别:
Direct
直接交换器,工作方式类似于单播,Exchange会将消息发送完全匹配ROUTING_KEY的Queue
fanout
广播是式交换器,不管消息的ROUTING_KEY设置为什么,Exchange都会将消息转发给所有绑定的Queue。
topic
主题交换器,工作方式类似于组播,Exchange会将消息转发和ROUTING_KEY匹配模式相同的所有队列,比如,ROUTING_KEY为user.stock的Message会转发给绑定匹配模式为 * .stock,user.stock, * . * 和#.user.stock.#的队列。( * 表是匹配一个任意词组,#表示匹配0个或多个词组)
headers
消息体的header匹配(ignore)
Binding
所谓绑定就是将一个特定的 Exchange 和一个特定的 Queue 绑定起来。Exchange 和Queue的绑定可以是多对多的关系。
virtual host
在rabbitmq server上可以创建多个虚拟的message broker,又叫做virtual hosts (vhosts)。每一个vhost本质上是一个mini-rabbitmq server,分别管理各自的exchange,和bindings。vhost相当于物理的server,可以为不同app提供边界隔离,使得应用安全的运行在不同的vhost实例上,相互之间不会干扰。producer和consumer连接rabbit server需要指定一个vhost。
1.3、消息队列的使用过程
1. 客户端连接到消息队列服务器,打开一个channel。
2. 客户端声明一个exchange,并设置相关属性。
3. 客户端声明一个queue,并设置相关属性。
4. 客户端使用routing key,在exchange和queue之间建立好绑定关系。
5. 客户端投递消息到exchange。
6. exchange接收到消息后,就根据消息的key和已经设置的binding,进行消息路由,将消息投递到一个或多个队列里
rabbitMq.xml配置信息
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 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/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">
<!--配置connection-factory,指定连接rabbit server参数 -->
<rabbit:connection-factory id="connectionFactory"
username="asdf" password="123456" host="10.75.4.25" port="5672" />
<!--定义rabbit template用于数据的接收和发送 -->
<rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
exchange="exchangeTest" />
<!--通过指定下面的admin信息,当前producer中的exchange和queue会在rabbitmq服务器上自动生成 -->
<rabbit:admin connection-factory="connectionFactory" />
<!--定义queue -->
<rabbit:queue name="queueTest" durable="true" auto-delete="false" exclusive="false" />
<!-- 定义direct exchange,绑定queueTest -->
<rabbit:direct-exchange name="exchangeTest" durable="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="queueTest" key="queueTestKey"></rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>
<!-- 消息接收者 -->
<bean id="messageReceiver" class="com.lin.consumer.MessageConsumer"></bean>
<!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener queues="queueTest" ref="messageReceiver"/>
</rabbit:listener-container>
</beans>
消息生产者
package com.lin.producer;
import javax.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;
/**
* 功能概要:消息产生,提交到队列中去
*
* @author linbingwen
* @since 2016年1月15日
*/
@Service
public class MessageProducer {
private Logger logger = LoggerFactory.getLogger(MessageProducer.class);
@Resource
private AmqpTemplate amqpTemplate;
public void sendMessage(Object message){
logger.info("to send message:{}",message);
amqpTemplate.convertAndSend("queueTestKey",message);
}
}
3、消息消费者
package com.lin.consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
/**
* 功能概要:消费接收
*
* @author linbingwen
* @since 2016年1月15日
*/
public class MessageConsumer implements MessageListener {
private Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
@Override
public void onMessage(Message message) {
logger.info("receive message:{}",message);
}
}
1.实现一个消息监听器ReceiveMessageListener.Java
[java] view plain copy
print?在CODE上查看代码片派生到我的代码片
package org.springframework.amqp.core;
/**
* Listener interface to receive asynchronous delivery of Amqp Messages.
*
* @author Mark Pollack
*/
public interface MessageListener {
void onMessage(Message message);
}
2.消费者配置Consumer.xml
[html] view plain copy
print?在CODE上查看代码片派生到我的代码片
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<!-- 连接服务配置 -->
<rabbit:connection-factory id="connectionFactory" host="192.168.36.102" username="admin"
password="admin" port="5672" virtual-host="/" channel-cache-size="5" />
<rabbit:admin connection-factory="connectionFactory"/>
<!-- queue 队列声明-->
<rabbit:queue durable="true" auto-delete="false" exclusive="false" name="spring.queue.tag"/>
<!-- exchange queue binging key 绑定 -->
<rabbit:direct-exchange name="spring.queue.exchange" durable="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="spring.queue.tag" key="spring.queue.tag.key"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<bean id="receiveMessageListener"
class="cn.slimsmart.rabbitmq.demo.spring.tag.ReceiveMessageListener" />
<!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" >
<rabbit:listener queues="spring.queue.tag" ref="receiveMessageListener" />
</rabbit:listener-container>
</beans>
3.生产者配置Producer.xml
[html] view plain copy
print?在CODE上查看代码片派生到我的代码片
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<!-- 连接服务配置 -->
<rabbit:connection-factory id="connectionFactory"
host="192.168.36.102" username="admin" password="admin" port="5672"
virtual-host="/" channel-cache-size="5" />
<rabbit:admin connection-factory="connectionFactory" />
<!-- queue 队列声明 -->
<rabbit:queue durable="true"
auto-delete="false" exclusive="false" name="spring.queue.tag" />
<!-- exchange queue binging key 绑定 -->
<rabbit:direct-exchange name="spring.queue.exchange"
durable="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="spring.queue.tag" key="spring.queue.tag.key" />
</rabbit:bindings>
</rabbit:direct-exchange>
<!-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列,由于Gson的速度快于jackson,这里替换为Gson的一个实现 -->
<bean id="jsonMessageConverter"
class="cn.slimsmart.rabbitmq.demo.spring.tag.Gson2JsonMessageConverter" />
<!-- spring template声明 -->
<rabbit:template id="amqpTemplate" exchange="spring.queue.exchange" routing-key="spring.queue.tag.key"
connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
</beans>
4.消费者启动类ConsumerMain.java
[java] view plain copy
print?在CODE上查看代码片派生到我的代码片
package cn.slimsmart.rabbitmq.demo.spring.tag;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ConsumerMain {
public static void main(String[] args) {
new ClassPathXmlApplicationContext("Consumer.xml");
}
}
5.生产者启动类ProducerMain.java
[java] view plain copy
print?在CODE上查看代码片派生到我的代码片
package cn.slimsmart.rabbitmq.demo.spring.tag;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ProducerMain {
public static void main(String[] args) {
ApplicationContext context = new ClassPathXmlApplicationContext("Producer.xml");
AmqpTemplate amqpTemplate = context.getBean(RabbitTemplate.class);
User user = new User();
user.setName("niuniu");
amqpTemplate.convertAndSend(user);
}
}
1.RabbitMQ简介
RabbitMQ是流行的开源消息队列系统,用erlang语言开发。RabbitMQ是AMQP(高级消息队列协议)的标准实现。
官网:http://www.rabbitmq.com/
2.Spring集成RabbitMQ
2.1 maven配置
//pom.xml
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.5.1</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.4.5.RELEASE</version>
</dependency>
2.2 rabbmitmq配置文件
//rabbitmq-config.properties
mq.host=127.0.0.1
mq.username=test
mq.password=123456
mq.port=5672
mq.vhost=testmq
2.3 Spring配置
//application-mq.xml
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
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/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" >
<description>rabbitmq 连接服务配置</description>
<!-- 连接配置 -->
<rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="${mq.port}" virtual-host="${mq.vhost}"/>
<rabbit:admin connection-factory="connectionFactory"/>
<!-- spring template声明-->
<rabbit:template exchange="amqpExchange" id="amqpTemplate" connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
<!-- 消息对象json转换类 -->
<bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
</beans>
3. 在Spring中使用RabbitMQ
3.1 申明一个消息队列Queue
//application-mq.xml
<rabbit:queue id="test_queue_key" name="test_queue_key" durable="true" auto-delete="false" exclusive="false" />
说明:
durable:是否持久化
exclusive: 仅创建者可以使用的私有队列,断开后自动删除
auto_delete: 当所有消费客户端连接断开后,是否自动删除队列
3.2 交换机定义
//application-mq.xml
<rabbit:direct-exchange name="test-mq-exchange" durable="true" auto-delete="false" id="test-mq-exchange">
<rabbit:bindings>
<rabbit:binding queue="test_queue_key" key="test_queue_key"/>
</rabbit:bindings>
</rabbit:direct-exchange>
说明:
rabbit:direct-exchange:定义exchange模式为direct,意思就是消息与一个特定的路由键完全匹配,才会转发。
rabbit:binding:设置消息queue匹配的key
3.3 发送消息Producer
//MQProducer.java
public interface MQProducer {
/**
* 发送消息到指定队列
* @param queueKey
* @param object
*/
public void sendDataToQueue(String queueKey, Object object);
}
@Service
public class MQProducerImpl implements MQProducer {
@Autowired
private AmqpTemplate amqpTemplate;
private final static Logger LOGGER = Logger.getLogger(MQProducerImpl.class);
/* (non-Javadoc)
* @see com.stnts.tita.rm.api.mq.MQProducer#sendDataToQueue(java.lang.String, java.lang.Object)
*/
@Override
public void sendDataToQueue(String queueKey, Object object) {
try {
amqpTemplate.convertAndSend(queueKey, object);
} catch (Exception e) {
LOGGER.error(e);
}
}
}
说明:
convertAndSend:将Java对象转换为消息发送到匹配Key的交换机中Exchange,由于配置了JSON转换,这里是将Java对象转换成JSON字符串的形式。原文:Convert a Java object to an Amqp Message and send it to a default exchange with a specific routing key.
3.4 异步接收消息Consumer
定义监听器
//QueueListenter.java
@Component
public class QueueListenter implements MessageListener {
@Override
public void onMessage(Message msg) {
try{
System.out.print(msg.toString());
}catch(Exception e){
e.printStackTrace();
}
}
}
监听配置
//application-mq.xml
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto">
<rabbit:listener queues="test_queue" ref="queueListenter"/>
</rabbit:listener-container>
说明:
queues:监听的队列,多个的话用逗号(,)分隔
ref:监听器
3.5 JUnit测试
//TestQueue.java
@RunWith(value = SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {
"classpath:/ApplicationContext/ApplicationContext-mq.xml"})
public class TestQueue{
@Autowired
MQProducer mqProducer;
final String queue_key = "test_queue";
@Test
public void send(){
Map<String,Object> msg = new HashMap()<>;
msg.put("data","hello,rabbmitmq!");
mqProducer.sendDataToQueue(query_key,msg);
}
}
Spring Rabbitmq使用实例
Message Producer
生产者的Rabbitmq.xml配置,这里我们使用rabbit标签来配置rabbitmq客户端。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
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/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">
<!--配置connection-factory,指定连接rabbit server参数-->
<rabbit:connection-factory id="connectionFactory" username="guest" password="guest"
host="localhost"
port="5672"
virtual-host="/"/>
<!--通过指定下面的admin信息,当前producer中的exchange和queue会在rabbitmq服务器上自动生成-->
<rabbit:admin connection-factory="connectionFactory"/>
<!--定义queue-->
<rabbit:queue id="com.mj.test" name="com.mj.test" durable="true" auto-delete="false" exclusive="false"/>
<!-- 定义direct exchange,绑定com.mj.test queue -->
<rabbit:direct-exchange name="myChange" durable="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="com.mj.test" key="hello"></rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>
<!--定义rabbit template用于数据的接收和发送-->
<rabbit:template id="myAmqpTemplate" connection-factory="connectionFactory" exchange="myChange"/>
</beans>
ApplicationContext.xml配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.1.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
http://www.springframework.org/schema/jdbc
http://www.springframework.org/schema/jdbc/spring-jdbc-3.1.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-3.1.xsd">
<!-- 使能AOP-->
<aop:aspectj-autoproxy/>
<!-- 自动装载bean使能-->
<context:component-scan base-package="com.mj.amq"/>
<context:annotation-config/>
<!-- 声明一个事物管理 -->
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"></property>
</bean>
<!-- 启用annotation的事务支持 -->
<tx:annotation-driven transaction-manager="transactionManager" proxy-target-class="true"/>
<import resource="Rabbitmq.xml"/>
<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://localhost:3307/student?useUnicode=true&characoding=UTF-8"/>
<property name="username" value="root"/>
<property name="password" value="root"/>
</bean>
</beans>
Producer发送数据Java代码
@Service
public class MessageSender {
@Resource(name="myAmqpTemplate")
AmqpTemplate amqpTemplate;
public void sendMessage(Object message){
amqpTemplate.convertAndSend("hello",message);
}
}
Message Consumer
Rabbitmq.xml配置
Rabbitmq接收端需要配置Connection-Factory实例,监听的队列,以及Listener-container。同时,我们需要创建监听队列的java bean。这里我们使用的是spring rabbitmq提供的异步MessageListener接口,consumer的业务逻辑在onMessage中实现。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
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/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">
<rabbit:connection-factory id="connectionFactory" username="guest" password="guest"
host="localhost"
port="5672"
virtual-host="/"/>
<rabbit:queue id="com.mj.test" name="com.mj.test" durable="true" auto-delete="false" exclusive="false"/>
<bean id="messageReceiver" class="com.mj.amq.MessageReceiver"></bean>
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener queues="com.mj.test" ref="messageReceiver"/>
</rabbit:listener-container>
</beans>
ApplicationContext.xml和producer的一致
Consumer侦听mq消息代码
public class MessageReceiver implements MessageListener {
public void onMessage(Message message) {
System.out.println(message);
}
public static void main(String[]args){
ApplicationContext applicationContext=new ClassPathXmlApplicationContext("ApplicationContext.xml");
}
}
测试
启动MessageReceiver的main函数,运行producer的单元测试,会在MessageReceiver的concole端看到接收都到的数据。同时登陆rabbitmq客户端可以看到我们配置的Exchange和queue信息。
public class MessageSenderTest {
private ApplicationContext context = null;
@Before
public void setUp() throws Exception {
context = new ClassPathXmlApplicationContext("ApplicationContext.xml");
}
@Test
public void should_send_a_amq_message() throws Exception {
MessageSender messageSender = (MessageSender) context.getBean("messageSender");
messageSender.sendMessage("Hello, I am amq sender");
}
}
相关推荐
SpringBoot 使用MQ
计算机技术、IT咨询、人工智能AI理论介绍,学习参考资料计算机技术、IT咨询、人工智能AI理论介绍,学习参考资料计算机技术、IT咨询、人工智能AI理论介绍,学习参考资料计算机技术、IT咨询、人工智能AI理论介绍,学习...
通过阅读《RabbitMQ总结.docx》这份文档,你将更深入地了解RabbitMQ的配置、使用、最佳实践以及常见问题的解决方法。文档可能涵盖了如何安装和启动RabbitMQ服务器,创建和管理虚拟主机,设置用户权限,以及如何在...
Windows RabbitMQ 安装总结 本文总结了 Windows 平台下 RabbitMQ 的安装过程,从下载 Erlang 到安装 RabbitMQ 服务器,并提供了 Java 代码示例,展示了生产者和消费者端的实现。 一、Erlang 的安装 Erlang 是 ...
### RabbitMQ 使用总结 #### 一、安装与配置 **1. 安装顺序** - **ERLANG安装**: 首先确保安装了Erlang环境。安装时注意选择带有`windowdll`选项的版本,这有助于后续RabbitMQ的稳定运行。 - **RabbitMQ安装**: ...
**正文** ...总结,`rabbitmq-c-master`的编译过程主要涉及CMake的使用,以及对RabbitMQ和AMQP协议的理解。通过这个过程,开发者可以将RabbitMQ的功能集成到C语言的应用中,实现可靠的消息传递和异步处理。
RabbitMQ是一款开源的...总结,安装RabbitMQ 3.7.8 on Windows涉及Erlang环境的准备、RabbitMQ服务的安装与配置,以及基本的管理和使用知识。正确配置和使用RabbitMQ,能有效提升分布式系统中的消息通信效率和可靠性。
总结来说,RabbitMQ与SpringMVC的集成可以极大地增强Web应用的异步处理能力,提高系统的可扩展性。在实际应用中,需要正确安装和配置Erlang及RabbitMQ,然后利用Spring的AMQP支持将消息队列功能融入到Web应用中,...
总结来说,`rabbitmq.config`和`advanced.config`是RabbitMQ服务器的核心配置文件,它们定义了RabbitMQ的行为和性能特性。理解并正确配置这些文件对于优化RabbitMQ的性能和稳定性至关重要,特别是在分布式环境中。...
总结,RabbitMQ作为消息中间件,为Java开发者提供了强大的工具,帮助构建高可用、可扩展的系统。"rabbitMQ-demo.zip"中的"rabbitMQ-demo-main"项目是学习和实践RabbitMQ的宝贵资源,通过它,我们可以更深入地理解...
总结来说,使用RabbitMQ在Android上进行网络发送涉及服务器安装、客户端集成、消息队列的声明和使用,以及AMQP协议的交互。对于初学者来说,理解这些基本概念和操作是至关重要的,可以帮助构建可靠和可扩展的跨平台...
总结一下,这个“RabbitMQ Java测试客户端”项目是一个实践示例,展示了如何在Java应用程序中使用RabbitMQ进行消息传递。它涵盖了连接管理、消息发送与接收的基本流程,对于学习和理解RabbitMQ在Java环境下的应用...
总结来说,Linux上的RabbitMQ安装需要先安装Erlang环境,然后安装RabbitMQ服务器,并进行必要的配置和安全管理。RabbitMQ作为一个强大的消息中间件,能够有效地帮助你在分布式系统中实现异步处理和解耦。在实际应用...
总结来说,RabbitMQ 的安装与 MQTT 功能的开启涉及下载和安装 Erlang OTP 以及 RabbitMQ 服务器,启用 MQTT 插件,配置参数,最后验证 MQTT 连接。这是一项关键的操作,使得 RabbitMQ 能够支持物联网设备和其他 MQTT...
总结来说,RabbitMQ自动安装脚本是高效部署RabbitMQ的关键工具,它涵盖了从环境准备、依赖安装到服务启动的全过程。理解这一过程有助于我们更好地管理和维护RabbitMQ服务器,特别是在离线环境中,确保服务的稳定性和...
RabbitMQ知识点总结,基本能解决项目需求
总结来说,RabbitMQ是一个强大的消息中间件,它的安装通常涉及下载安装包、解压、安装和启动服务。了解RabbitMQ的基本概念和使用方式对于构建高效、可靠的分布式系统至关重要。通过熟练掌握RabbitMQ,开发者可以构建...
总结起来,这个C# Demo项目旨在教会我们如何在.NET环境中使用RabbitMQ进行消息传递,通过封装RabbitMQ.Client库,我们可以更高效地利用交换机功能,实现不同场景下的消息路由。通过学习这个项目,开发者能够更好地...
【RabbitMQ】是著名的开源消息中间件,它基于AMQP(Advanced Message Queuing Protocol)协议实现,提供了高效、可信赖的消息传递服务。RabbitMQ的使用能够解决多个IT领域中的核心问题,如异步处理、应用解耦、流量...
总结来说,"RabbitMQ.tar.gz"压缩包提供了一站式的解决方案,帮助用户在CentOS 7环境下快速搭建和运行RabbitMQ服务器。通过包含Erlang环境和RabbitMQ服务器,用户可以轻松地利用消息队列技术提升系统的可扩展性和...