RabbitMQ topic模式
生产者
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-exchangeTopic" connection-factory="rabbitConnectionFactory"/>
<!-- 申明exchange -->
<rabbit:topic-exchange id="myTest-exchangeTopic" name="myTest-exchangeTopic" durable="true" auto-declare="true" auto-delete="false">
</rabbit:topic-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("zjs.com.AS1", "消息a1");
amqpTemplate.convertAndSend("zjs.com.AS2", "消息a2");
amqpTemplate.convertAndSend("zjs.com.AS3", "消息a3");
amqpTemplate.convertAndSend("zjs.cn.BS1", "消息b1");
amqpTemplate.convertAndSend("zjs.cn.BS2", "消息b2");
amqpTemplate.convertAndSend("zjs.cn.BS3", "消息b3");
System.out.println("发送消息成功");
return "Hello";
}
}
消费者
- 配置文件
applicationContext-springmq.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.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="topicQueuex" name="topicQueuex" durable="true" auto-declare="true" auto-delete="false"/>
<rabbit:queue id="topicQueueb" name="topicQueueb" 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="topicQueuex" method="onMessage" ref="consumeraListener"/>
<rabbit:listener queues="topicQueueb" method="onMessage" ref="consumerbListener"/>
</rabbit:listener-container>
<rabbit:topic-exchange id="myTest-exchangeTopic" name="myTest-exchangeTopic" durable="true" auto-declare="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="topicQueuex" pattern="zjs.com.AS1"/>
<rabbit:binding queue="topicQueueb" pattern="zjs.cn.BS1"/>
</rabbit:bindings>
</rabbit:topic-exchange>
</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
总结
1.一般在topic模式中
生产端只申明不绑定，由消费端去绑定
amqpTemplate.convertAndSend("zjs.com.AS1", "消息a1");
amqpTemplate.convertAndSend("zjs.com.AS2", "消息a2");
amqpTemplate.convertAndSend("zjs.com.AS3", "消息a3");
amqpTemplate.convertAndSend("zjs.cn.BS1", "消息b1");
amqpTemplate.convertAndSend("zjs.cn.BS2", "消息b2");
amqpTemplate.convertAndSend("zjs.cn.BS3", "消息b3");
生产端配置：
<!-- 申明exchange -->
<rabbit:topic-exchange id="myTest-exchangeTopic" name="myTest-exchangeTopic" durable="true" auto-declare="true" auto-delete="false">
</rabbit:topic-exchange>
消费端配置：
<rabbit:topic-exchange id="myTest-exchangeTopic" name="myTest-exchangeTopic" durable="true" auto-declare="true" auto-delete="false">
<rabbit:bindings>
<rabbit:binding queue="topicQueuex" pattern="zjs.com.AS1"/>
<rabbit:binding queue="topicQueueb" pattern="zjs.*.*"/>
</rabbit:bindings>
</rabbit:topic-exchange>
2.不管是生产端还是消费端，申明的exchange或者queue，如果名称相同的对象都是在rabbit服务器的同一个实例。
发送消息成功
BListener接收到消息：消息a1
AListener接收到消息：消息a1
BListener接收到消息：消息a2
BListener接收到消息：消息a3
BListener接收到消息：消息b1
BListener接收到消息：消息b2
BListener接收到消息：消息b3
