RabbitMQ direct模式
关于它的快速入门，可以在这里查看：http://www.rabbitmq.com/getstarted.html。
生产者
1.配置文件
<?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.1.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.3.xsd">
<!-- 连接服务配置 -->
<rabbit:connection-factory id="rabbitConnectionFactory" addresses="${base-mq.addresses}"
virtual-host="${base-mq.vhost}" username="${base-mq.username}"
password="${base-mq.password}"/>
<rabbit:admin connection-factory="rabbitConnectionFactory"/>
<!-- template申明 -->
<rabbit:template id="amqpTemplate" exchange="myTest-exchangeFirst" connection-factory="rabbitConnectionFactory"/>
<!-- 队列 -->
<rabbit:queue id="testQueuea" name="testQueuea" durable="true" auto-declare="true" auto-delete="false"/>
<rabbit:queue id="testQueueb" name="testQueueb" durable="true" auto-declare="true" auto-delete="false"/>
<!-- 申明exchange -->
<rabbit:direct-exchange id="myTest-exchangeFirst" name="myTest-exchangeFirst" durable="true" auto-declare="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="testQueuea" key="keya"/>
<rabbit:binding queue="testQueueb" key="keyb"/>
<rabbit:binding queue="testQueueb" key="keyab"/>
</rabbit:bindings>
</rabbit:direct-exchange>
</beans>
- 生产者代码
package com.zjs.sp.rabbit.web.controller;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
@Controller
@RequestMapping(value = "/attribute")
public class RabbitProducerController {
@Autowired
private AmqpTemplate amqpTemplate;
@RequestMapping("/hello")
public String hello() {
amqpTemplate.convertAndSend("keya", "消息a1");
amqpTemplate.convertAndSend("keya", "消息a2");
amqpTemplate.convertAndSend("keya", "消息a3");
amqpTemplate.convertAndSend("keyb", "消息b1");
amqpTemplate.convertAndSend("keyb", "消息b2");
amqpTemplate.convertAndSend("keyb", "消息b3");
amqpTemplate.convertAndSend("keyab", "消息ab");
System.out.println("发送消息成功");
return "Hello";
}
}
消费者
- 配置文件
applicationContext-springmq.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.1.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.3.xsd">
<!-- 连接配置 -->
<rabbit:connection-factory id="rabbitConnectionFactory" addresses="${base-mq.addresses}"
virtual-host="${base-mq.vhost}" username="${base-mq.username}" password="${base-mq.password}"/>
<rabbit:admin connection-factory="rabbitConnectionFactory"/>
<!-- 队列 -->
<rabbit:queue id="testQueuea" name="testQueuea" durable="true" auto-declare="true" auto-delete="false"/>
<rabbit:queue id="testQueueb" name="testQueueb" durable="true" auto-declare="true" auto-delete="false"/>
<!-- 监听类 -->
<bean id="consumeraListener" class="com.zjs.sp.consumer.web.listener.ConsumeraListener"/>
<bean id="consumerbListener" class="com.zjs.sp.consumer.web.listener.ConsumerbListener"/>
<!-- 消费queue的listener -->
<rabbit:listener-container connection-factory="rabbitConnectionFactory" acknowledge="auto" concurrency="1" >
<rabbit:listener queues="testQueuea" method="onMessage" ref="consumeraListener"/>
<rabbit:listener queues="testQueueb" method="onMessage" ref="consumerbListener"/>
</rabbit:listener-container>
</beans>
- Listener代码
ConsumeraListener .java
package com.zjs.sp.consumer.web.listener;
/**
* @author zhangjinsong
*/
public class ConsumeraListener {
public void onMessage(String message) {
System.out.println("AListener接收到消息："+message);
}
}
ConsumerbListener .java
package com.zjs.sp.consumer.web.listener;
/**
* @author zhangjinsong
*/
public class ConsumerbListener {
public void onMessage(String message) {
System.out.println("BListener接收到消息："+message);
}
}
执行结果
发送消息成功
AListener接收到消息：消息a1
BListener接收到消息：消息b1
AListener接收到消息：消息a2
BListener接收到消息：消息b2
AListener接收到消息：消息a3
BListener接收到消息：消息b3
BListener接收到消息：消息ab
RabbitMQ Management绑定关系如下
image.png
总结
1.一般在direct模式中生产端申明绑定，由消费端只负责监听
amqpTemplate.convertAndSend("keya", "消息a1");
amqpTemplate.convertAndSend("keya", "消息a2");
amqpTemplate.convertAndSend("keya", "消息a3");
amqpTemplate.convertAndSend("keyb", "消息b1");
amqpTemplate.convertAndSend("keyb", "消息b2");
amqpTemplate.convertAndSend("keyb", "消息b3");
amqpTemplate.convertAndSend("keyab", "消息ab");
生产端配置：
<!-- 申明exchange -->
<rabbit:direct-exchange id="myTest-exchangeFirst" name="myTest-exchangeFirst" durable="true" auto-declare="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="testQueuea" key="keya"/>
<rabbit:binding queue="testQueueb" key="keyb"/>
<rabbit:binding queue="testQueueb" key="keyab"/>
</rabbit:bindings>
</rabbit:direct-exchange>
消费端配置：
<!-- 队列 -->
<rabbit:queue id="testQueuea" name="testQueuea" durable="true" auto-declare="true" auto-delete="false"/>
<rabbit:queue id="testQueueb" name="testQueueb" durable="true" auto-declare="true" auto-delete="false"/>
<rabbit:queue id="testQueuec" name="testQueuec" durable="true" auto-declare="true" auto-delete="false"/>
<!-- 监听类 -->
<bean id="consumeraListener" class="com.zjs.sp.consumer.web.listener.ConsumeraListener"/>
<bean id="consumerbListener" class="com.zjs.sp.consumer.web.listener.ConsumerbListener"/>
<bean id="consumercListener" class="com.zjs.sp.consumer.web.listener.ConsumercListener"/>
<!-- 消费queue的listener -->
<rabbit:listener-container connection-factory="rabbitConnectionFactory" acknowledge="auto" concurrency="1" >
<rabbit:listener queues="testQueuea" method="onMessage" ref="consumeraListener"/>
<rabbit:listener queues="testQueueb" method="onMessage" ref="consumerbListener"/>
<rabbit:listener queues="testQueuec" method="onMessage" ref="consumercListener"/>
</rabbit:listener-container>
2.不管是生产端还是消费端，申明的exchange或者queue，如果名称相同的对象都是在rabbit服务器的同一个实例。
当然消费端同样可以再申明exchange ，如果名称一致不会重新创建，如果有新的绑定可以更新上去，例如消费端：
<!-- 队列 -->
<rabbit:queue id="testQueuea" name="testQueuea" durable="true" auto-declare="true" auto-delete="false"/>
<rabbit:queue id="testQueueb" name="testQueueb" durable="true" auto-declare="true" auto-delete="false"/>
<rabbit:queue id="testQueuec" name="testQueuec" durable="true" auto-declare="true" auto-delete="false"/>
<!-- 监听类 -->
<bean id="consumeraListener" class="com.zjs.sp.consumer.web.listener.ConsumeraListener"/>
<bean id="consumerbListener" class="com.zjs.sp.consumer.web.listener.ConsumerbListener"/>
<bean id="consumercListener" class="com.zjs.sp.consumer.web.listener.ConsumercListener"/>
<!-- 消费queue的listener -->
<rabbit:listener-container connection-factory="rabbitConnectionFactory" acknowledge="auto" concurrency="1" >
<rabbit:listener queues="testQueuea" method="onMessage" ref="consumeraListener"/>
<rabbit:listener queues="testQueueb" method="onMessage" ref="consumerbListener"/>
<rabbit:listener queues="testQueuec" method="onMessage" ref="consumercListener"/>
</rabbit:listener-container>
<!--绑定 -->
<rabbit:direct-exchange id="myTest-exchangeFirst" name="myTest-exchangeFirst" durable="true" auto-declare="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="testQueuec" key="keyab"/>
</rabbit:bindings>
</rabbit:direct-exchange>
RabbitMQ Management绑定关系如下
image.png
执行结果如下
发送消息成功
BListener接收到消息：消息b1
CListener接收到消息：消息ab
AListener接收到消息：消息a1
BListener接收到消息：消息b2
AListener接收到消息：消息a2
BListener接收到消息：消息b3
AListener接收到消息：消息a3
BListener接收到消息：消息ab
