RabbitMQ topic模式

生產(chǎn)者

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">
    
    <!-- 連接服務(wù)配置 -->
    <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>

2. 生產(chǎn)者代碼

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("發(fā)送消息成功");
        return "Hello";
    }
}

消費(fèi)者

1. 配置文件
   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"/>
    
    <!-- 隊(duì)列 -->
    <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"/>
    <!-- 監(jiān)聽類 -->
    <bean id="consumeraListener" class="com.zjs.sp.consumer.web.listener.ConsumeraListener"/>
    <bean id="consumerbListener" class="com.zjs.sp.consumer.web.listener.ConsumerbListener"/>
    <!--     消費(fèi)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>

2. 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);
    }
}

執(zhí)行結(jié)果

發(fā)送消息成功
AListener接收到消息：消息a1
BListener接收到消息：消息b1

總結(jié)

1.一般在topic模式中
生產(chǎn)端只申明不綁定，由消費(fèi)端去綁定

  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");

生產(chǎn)端配置：

    <!-- 申明exchange -->
    <rabbit:topic-exchange id="myTest-exchangeTopic" name="myTest-exchangeTopic" durable="true" auto-declare="true" auto-delete="false">
    </rabbit:topic-exchange>

消費(fèi)端配置：

    <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.不管是生產(chǎn)端還是消費(fèi)端，申明的exchange或者queue，如果名稱相同的對象都是在rabbit服務(wù)器的同一個(gè)實(shí)例。

發(fā)送消息成功
BListener接收到消息：消息a1
AListener接收到消息：消息a1
BListener接收到消息：消息a2
BListener接收到消息：消息a3
BListener接收到消息：消息b1
BListener接收到消息：消息b2
BListener接收到消息：消息b3