RabbitMQ fanout模式

生产者

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-exchangeFanout"  connection-factory="rabbitConnectionFactory"/>
    
    <!-- 申明exchange -->
    <rabbit:fanout-exchange id="myTest-exchangeFanout" name="myTest-exchangeFanout" durable="true" auto-declare="true" auto-delete="false">
    </rabbit:fanout-exchange>
</beans>

2. 生产者代码

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("消息a1");
        amqpTemplate.convertAndSend("消息a2");
        amqpTemplate.convertAndSend("消息a3");

        amqpTemplate.convertAndSend("消息b1");
        amqpTemplate.convertAndSend("消息b2");
        amqpTemplate.convertAndSend("消息b3");

        amqpTemplate.convertAndSend("消息ab");
        System.out.println("发送消息成功");
        return "Hello";
    }
}

消费者

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"/>
    
    <!-- 队列 -->
    <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>
    
    <!--绑定 -->
    <rabbit:fanout-exchange id="myTest-exchangeFanout" name="myTest-exchangeFanout" durable="true" auto-declare="true" auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding queue="testQueuea"/>
            <rabbit:binding queue="testQueueb"/>
        </rabbit:bindings>
    </rabbit:fanout-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);
    }
}

执行结果

发送消息成功
AListener接收到消息：消息a1
BListener接收到消息：消息a1
BListener接收到消息：消息a2
AListener接收到消息：消息a2
BListener接收到消息：消息a3
AListener接收到消息：消息a3
BListener接收到消息：消息b1
AListener接收到消息：消息b1
BListener接收到消息：消息b2
AListener接收到消息：消息b2
BListener接收到消息：消息b3
AListener接收到消息：消息b3
BListener接收到消息：消息ab
AListener接收到消息：消息ab