Spring Boot 整合 RabbitMQ 之 Fanout Exchange模式 (三)

摘要：那前面已经介绍过了Direct模式 (一)Topic转发模式 (二)，这次介绍下Fanout Exchange形式又叫广播形式,因此我们发送到路由器的消息会使得绑定到该路由器的每一个Queue接收到消息,这个时候就算指定了Key,或者规则(即上文中convertAndSend方法的参数2),也会被忽略!那么直接上代码,发送端配置如下:

package com.micai.springboot.mq.config;

import com.micai.springboot.base.BaseConfig;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 描述：Topic转发模式
 * <p>
 * Author: 赵新国
 * Date: 2017/11/3 18:51
 */
@Configuration
public class SenderConf extends BaseConfig {

    // ---------------------------------------------------- Direct 形式 -------------------------------------------- //

    /*@Bean
    public Queue queue() {
        return new Queue(QUEUE_KEY);
    }*/

    // ---------------------------------------------------- Topic 形式 -------------------------------------------- //

    /*@Bean(name = "message")
    public Queue queueMessage() {
        return new Queue("topic.message");
    }

    @Bean(name = "messages")
    public Queue queueMessages() {
        return new Queue("topic.messages");
    }

    @Bean
    public TopicExchange exchange() {
        return new TopicExchange("exchange");
    }

    @Bean
    Binding bindingExchangeMessage(@Qualifier("message") Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    @Bean
    Binding bindingExchangeMessages(@Qualifier("messages") Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");//*表示一个词,#表示零个或多个词
    }*/

    // -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //

    @Bean(name="Amessage")
    public Queue AMessage() {
        return new Queue("fanout.A");
    }


    @Bean(name="Bmessage")
    public Queue BMessage() {
        return new Queue("fanout.B");
    }

    @Bean(name="Cmessage")
    public Queue CMessage() {
        return new Queue("fanout.C");
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");//配置广播路由器
    }

    @Bean
    Binding bindingExchangeA(@Qualifier("Amessage") Queue AMessage,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(@Qualifier("Bmessage") Queue BMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeC(@Qualifier("Cmessage") Queue CMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(CMessage).to(fanoutExchange);
    }

}

发送端使用如下代码发送:

package com.micai.springboot.mq;

import com.micai.springboot.base.BaseConfig;
import com.micai.springboot.entity.User;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 描述：消息生产者
 * <p>
 * Author: 赵新国
 * Date: 2017/11/3 15:37
 */
@Component
public class Sender extends BaseConfig {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        // ---------------------------------------------------- Direct 形式 -------------------------------------------- //
        // 在该生产者，我们会产生一个字符串，并发送到名为hello的队列中
        /*String context = "Hello " + "Rabbit MQ！";
        System.out.println("发送MQ消息 : " + context);
        this.rabbitTemplate.convertAndSend(QUEUE_KEY, context);*/

        // 发送对象，但是该对象必须实现Serializable接口
        /*User user = new User(); //实现Serializable接口
        user.setId(1L);
        user.setName("张三");
        this.rabbitTemplate.convertAndSend(QUEUE_KEY, user);*/

        // ---------------------------------------------------- Topic 形式 -------------------------------------------- //
        /*this.rabbitTemplate.convertAndSend("exchange", "topic.message", "hello, rabbit!");*/

        // -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
        this.rabbitTemplate.convertAndSend("fanoutExchange", "", "xixi,hlhdidi");// 参数2将被忽略
    }

}

接收端监听器配置如下:

package com.micai.springboot.mq;

import com.micai.springboot.base.BaseConfig;
import com.micai.springboot.entity.User;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 描述：消息消费者
 * @RabbitListener注解定义该类对hello队列的监听，
 * 并用@RabbitHandler注解来指定对消息的处理方法。
 * 所以，该消费者实现了对hello队列的消费，消费操作为输出消息的字符串内容
 * Author: 赵新国
 * Date: 2017/11/3 15:42
 */
@Component
public class Receiver extends BaseConfig {

    // ---------------------------------------------------- Direct 形式 -------------------------------------------- //
    //监听器监听指定的Queue
    /*@RabbitListener(queues = QUEUE_KEY)
    public void process(String str) {
        System.out.println("接收MQ消息 : " + str);
    }*/

    // //监听器监听指定的Queue
    /*@RabbitListener(queues = QUEUE_KEY)
    public void process(User user) { //用User作为参数
        System.out.println("接收MQ消息 : " + user);
    }*/

    // ---------------------------------------------------- Topic 形式 -------------------------------------------- //
    /*@RabbitListener(queues="topic.message")    //监听器监听指定的Queue
    public void process1(String str) {
        System.out.println("message:"+str);
    }

    @RabbitListener(queues="topic.messages")    //监听器监听指定的Queue
    public void process2(String str) {
        System.out.println("messages:"+str);
    }*/


    // -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
    @RabbitListener(queues="fanout.A")
    public void processA(String str1) {
        System.out.println("ReceiveA:"+str1);
    }

    @RabbitListener(queues="fanout.B")
    public void processB(String str) {
        System.out.println("ReceiveB:"+str);
    }

    @RabbitListener(queues="fanout.C")
    public void processC(String str) {
        System.out.println("ReceiveC:"+str);
    }

}

运行测试代码,发现三个监听器都接收到了数据,测试成功，结果如下图：

源代码下载地址：

​​https://gitee.com/micai/micai-springboot/tree/master/micai-springboot-rabbitmq-7​​