中间件系列四 RabbitMQ之Fanout exchange（扇型交换机）之发布订阅

╰半橙微兮° 2021-09-14 15:06 370阅读 0赞

## 1. 概述 ##

本文通过Fanout exchange（扇型交换机）实现生产者发送一个消息，这个消息同时被传送给所有消费者。Fanout exchange（扇型交换机）的定义见这篇文章[ 中间件系列三 RabbitMQ之交换机的四种类型和属性][_ RabbitMQ]

本文主要内容：

*  实现上文定义的场景
 *  交换机方法的声明方法、参数说明和实现
 *  临时队列的作用、声明方法和实现
 *  绑定(Bindings)的方法、参数说明和实现
 *  测试

下文先分别介绍生产者和消费者的代码，然后对其进行测试，模拟上面的发布/订阅场景。完整的代码在文末

## 2. 生产者代码 ##

主要逻辑在这个类中: Publish.java

1. 配置连接工厂
    2. 建立TCP连接
    3. 在TCP连接的基础上创建通道
    4. 声明一个fanout交换机 (而不是声明一个队列)
    5. 发送消息

这里和之前文章最大的不同是，多了声明一个fanout交换机，而不是声明队列。这里表明我们不使用默认交换机，在消费端需要执行队列和交换机的绑定操作

**声明交换机方法**

Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,
                                           Map<String, Object> arguments) throws IOException

详细参数如下：

> **第一个参数exchange**：交换机的名称  
> **第二个参数type**：交换机的类型  
> **第三个参数durable**：是否持久化，如果true，则当前RabbitMQ重启的时候，它依旧存在  
> **第四个参数autoDelete**：当没有生成者/消费者使用此交换机时，此交换机会被自动删除。  
> **第五个参数arguments**：其它的扩展属性

**代码如下**

// 声明一个fanout交换机
    channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.FANOUT, false, false, null);
    // 发送消息
    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));

**完整代码**  
发送者代码: [Publish.java][]

## 3. 消费者代码 ##

消费者主要逻辑在这个类中: Publish.java

1. 配置连接工厂
    2. 建立TCP连接
    3. 在TCP连接的基础上创建通道
    4. 声明一个fanout交换机 
    5. 声明一个临时队列
    6. 将临时队列绑定到交换机上
    7. 接收消息并处理

**临时队列：**  
在本例子中，我们使用临时队列。临时队列有以下特点：  
1. 消费者每次连接时，都会创建一个新的队列，队列名称随机生成  
2. 当消费者断开连接时，队列会自动变删除

在RabbitMQ中，随机使用的队列名称类似amq.gen-**\*\***

声明方法:

Queue.DeclareOk queueDeclare() throws IOException;  
    
    此方法等价于创建一个
    Queue.DeclareOk queueDeclare(String queue="自动生成类似amq.gen-******的随机名称", boolean durable=false, boolean exclusive=true, boolean autoDelete=true,  Map<String, Object> arguments=null) throws IOException;

代码：创建临时队列，得到队列名称，用于后面的绑定使用

String queueName = channel.queueDeclare().getQueue();

**绑定(Bindings)**  
![这里写图片描述][SouthEast]

在创建交换机、队列后，要实现将发送到特定的交换机X的消息路由到我们创建的队列，还需要做绑定操作。如果绑定成功后，交换机会将带特定路由键的消息路由到绑定的队列

方法：将队列绑定到交换机上

Queue.BindOk queueBind(String queue, String exchange, String routingKey) throws IOException;

详细参数如下：

> **第一个参数queue**：要绑定的队列  
> **第二个参数exchange**：要绑定的交换机  
> **第三个参数routingKey**：绑定使用的路由键

代码：将队列绑定到交换机上，因为我们这个demo是使用扇形交换机，routingKey是没有意义的，所以这里使用的routingKey为空字符串

String queueName = channel.queueDeclare().getQueue();
    channel.queueBind(queueName, EXCHANGE_NAME, "");

**完整代码**  
消费者代码：[Subscribe.java][]

## 4. 测试 ##

测试代码

public class BasicTest { 
        // 测试线程池
        private ExecutorService executorService = Executors.newFixedThreadPool(10);
    
        // rabbitmq的IP地址
        private final String rabbitmq_host = "10.240.80.147";
        // rabbitmq的用户名称
        private final String rabbitmq_user = "hry";
        // rabbitmq的用户密码
        private final String rabbitmq_pwd = "hry";
    
        @Test
        public void publishsubscribe() throws InterruptedException {
            // 接收端
            int recNum = 2;
            while(recNum-- > 0) {
                final int recId = recNum;
                executorService.submit(() -> {
                    Subscribe.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd, recId);
                });
            }
            Thread.sleep(5* 100);
            // 发送端
            int sendNum = 2;
            while(sendNum-- > 0){
                executorService.submit(() -> {
                    Publish.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd);
                });
            }
    
            // sleep 10s
            Thread.sleep(10 * 1000);
        }

执行测试，启动两个消费者，发送者发送2条记录，这些记录同时被2个消费者消费

[Subscribe-0] Waiting for messages.
     [Subscribe-1] Waiting for messages.
     // 发送两条记录
     [Publish] Sent 'Publish-366028801770803'
     [Publish] Sent 'Publish-366028803809517'
     // 消费者1收到两条记录
     [Subscribe-0] Received 'Publish-366028801770803'
     [Subscribe-0] Received 'Publish-366028803809517'
     // 消费者2收到两条记录
     [Subscribe-1] Received 'Publish-366028801770803'
     [Subscribe-1] Received 'Publish-366028803809517'

## 5.代码 ##

上文的详细代码主要如下：  
发送者代码: [Publish.java][]  
消费者代码：[Subscribe.java][]  
测试代码：[BasicTest.java][]的方法 publishsubscribe()  
所有的详细代码见[github代码，请尽量使用tag v0.8，不要使用master，因为master一直在变，不能保证文章中代码和github上的代码一直相同][github_tag v0.8_master_master_github]

[_ RabbitMQ]: http://blog.csdn.net/hry2015/article/details/79118804
[Publish.java]: https://github.com/hryou0922/spring_boot/blob/v0.8/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/publishsubscribe/Publish.java
[SouthEast]: /images/20210520/6443c9ff164a49808b5ec23329d5930c.png
[Subscribe.java]: https://github.com/hryou0922/spring_boot/blob/v0.8/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/publishsubscribe/Subscribe.java
[BasicTest.java]: https://github.com/hryou0922/spring_boot/blob/v0.8/rabbitmq/src/test/java/com/hry/spring/rabbitmq/basic/BasicTest.java
[github_tag v0.8_master_master_github]: https://github.com/hryou0922/spring_boot/tree/v0.8/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/publishsubscribe