spring boot 中使用 RabbitMQ 教程三 采用F

我们使用Fanout（扇出）模式，传递一个消息到多个消费者中，这个模式我们都听过就是发布与订阅。它的意思就是，发布的消息将被广播给所有接受者。

交换器

在前面的教程中我们向一个队列中发送和接收消息。这次介绍rabbitmq的全消息模式。

这里来快速回顾下以前的教程：

1. 一个生产者是发送消息的用户应用程序。
2. 一个队列是一个存储消息的缓冲区。
3. 一个消费是一个接收消息的用户应用程序。

在rabbitmq中消息传递的核心思想是，生产者永远不会直接的发送一个消息到队列。实际上生产者甚至都不知道，消息是否会被传递到任何一个队列中。

相反，生产者只能发送消息到一个交换器上。交换器是一个很简单的玩意。一头儿接收生产者发送的信息，另一头儿把消息推送到队列中。交换器必须明确的知道，如何处理接收到的消息。应该把它添加到指定的队列？还是添加到多个队列？或者给销毁？这些规则交给交换器定义。

rabbitmq一共有4个可用的交换器类型：direct, topic, headers 和fanout。 默认交换器是direct。发布与订阅的实现使用第四个交换器类型fanout。

具体实现

• Config.java 首先配置FanoutExchange为一个bean。然后定义2个队列。再之后定义2个Binding把2个队列绑定到FanoutExchange交换器上。 AnonymousQueue这个类是代表一个匿名的、不耐用的、独占的、自动删除的队列。Binding这个类是一个简单的容器，用来搜集信息并描述绑定。

package com.zb.rabbitMQtest.t3publishSubscribe.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author 张博
 */
@Configuration
public class Config {

    /**
     * 创建人：张博
     * 时间：2018/3/5 上午10:45
     * @apiNote 定义扇出（广播）交换器
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanout-exchange");
    }

    /**
     * 创建人：张博
     * 时间：2018/3/5 上午10:48
     * @apiNote 定义自动删除匿名队列
     */
    @Bean
    public Queue autoDeleteQueue0() {
        return new AnonymousQueue();
    }

    /**
     * 创建人：张博
     * 时间：2018/3/5 上午10:48
     * @apiNote 定义自动删除匿名队列
     */
    @Bean
    public Queue autoDeleteQueue1() {
        return new AnonymousQueue();
    }

    /**
     * 创建人：张博
     * 时间：2018/3/5 上午10:48
     * @param fanoutExchange 扇出（广播）交换器
     * @param autoDeleteQueue0 自动删除队列
     * @apiNote 把队列绑定到扇出（广播）交换器
     * @return Binding
     */
    @Bean
    public Binding binding0(FanoutExchange fanoutExchange, Queue autoDeleteQueue0) {
        return BindingBuilder.bind(autoDeleteQueue0).to(fanoutExchange);
    }
    
    /**
     * 创建人：张博
     * 时间：2018/3/5 上午10:55
     * @param fanoutExchange 扇出（广播）交换器
     * @param autoDeleteQueue1 自动删除队列
     * @apiNote 把队列绑定到扇出（广播）交换器
     * @return Binding
     */
    @Bean
    public Binding binding1(FanoutExchange fanoutExchange, Queue autoDeleteQueue1) {
        return BindingBuilder.bind(autoDeleteQueue1).to(fanoutExchange);
    }
}

• Receiver.java 接收广播的消息

package com.zb.rabbitMQtest.t3publishSubscribe.receiver;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author 张博
 */
@Component
public class Receiver {

    @RabbitListener(queues = "#{autoDeleteQueue0.name}")
    public void receiver0(String str) {
        System.out.println("receiver0++++++++++:" + str);
    }

    @RabbitListener(queues = "#{autoDeleteQueue1.name}")
    public void receiver1(String str) {
        System.out.println("receiver1++++++++++:" + str);
    }
}

• Send.java 广播消息

package com.zb.rabbitMQtest.t3publishSubscribe.send;

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author 张博
 */
@Component
public class Send {

    @Autowired
    private FanoutExchange fanoutExchange;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send() {
        for (int i = 0; i < 5; i++) {
            rabbitTemplate.convertAndSend(fanoutExchange.getName(), "", "哈哈哈");
        }
    }
}

• SendTest.java 测试

package com.zb.rabbitMQtest.t3publishSubscribe.send;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * @author 张博
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class SendTest {

    @Autowired
    private Send send;

    @Test
    public void send() throws Exception {
        send.send();
    }
}

• 运行结果

send++++++++++:哈哈哈
send++++++++++:哈哈哈
send++++++++++:哈哈哈
send++++++++++:哈哈哈
send++++++++++:哈哈哈

receiver0++++++++++:哈哈哈
receiver1++++++++++:哈哈哈
receiver1++++++++++:哈哈哈
receiver0++++++++++:哈哈哈
receiver1++++++++++:哈哈哈
receiver0++++++++++:哈哈哈
receiver1++++++++++:哈哈哈
receiver0++++++++++:哈哈哈
receiver1++++++++++:哈哈哈
receiver0++++++++++:哈哈哈

为什么在发布与订阅中使用的是AnonymousQueue队列？

1. 无论何时当我们链接rabbitmq时我们需要新的、空的队列。使用AnonymousQueue将创建一个带有随机名字的队列或者甚至更好的选择是让服务端为我们选择一个随机队列名给我们。

2. 一旦我们客户端断开链接，队列应该自动被删除。spring-amqp客户端为我们做到这点，它定义了一个AnonymousQueue，它是匿名的、不耐用的、独占的、自动删除的队列。

鉴于以上2点所以发布与订阅时采用AnonymousQueue是正确的选择。