RabbitMQ六种通信模式介绍——发布与订阅模式（Publish/Subscribe）

亦凉 2022-12-27 12:48 347阅读 0赞

### 文章目录 ###

*   *  一。简介
     *  二。代码实现
     *  三。程序运行效果
     *  四。模式总结

--------------------

## 一。简介 ##

本篇博客所讲的为RabbitMQ六种通信模式之一的发布与订阅模式，官网给出的图如下所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70]  
简单模式与工作模式2个案例中，只有3个角色：

**P**：生产者，也就是要发送消息的程序  
**C**：消费者：消息的接受者，会一直等待消息到来。  
**queue**：消息队列，图中红色部分

而在订阅模型中，多了一个exchange角色，而且过程略有变化：

**P**：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）  
**C**：消费者，消息的接受者，会一直等待消息到来。  
**Queue**：消息队列，接收消息、缓存消息。  
**Exchange**：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：

*  `Fanout`：广播，将消息交给所有绑定到交换机的队列
 *  `Direct`：定向，把消息交给符合指定routing key 的队列
 *  `Topic`：通配符，把消息交给符合routing pattern（路由模式） 的队列

**特别注意：Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！**

**`发布订阅模式：`**

1.  每个消费者监听自己的队列。
2.  生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息

## 二。代码实现 ##

1.连接工具类：

import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    /** * 创建连接工具类 */
    public class ConnectionUtil { 
    
        public static Connection getConnection() throws Exception { 
            //创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //主机地址;默认为 localhost
            connectionFactory.setHost("localhost");
            //连接端口;默认为 5672
            connectionFactory.setPort(5672);
            //虚拟主机名称;默认为 /
            connectionFactory.setVirtualHost("hello");
            //连接用户名；默认为guest
            connectionFactory.setUsername("guest");
            //连接密码；默认为guest
            connectionFactory.setPassword("guest");
    
            //创建连接
            return connectionFactory.newConnection();
        }
    
    }

2.消息生产者：

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * 发布与订阅使用的交换机类型为：fanout */
    public class Producer { 
    
       //交换机名称
       static final String FANOUT_EXCHAGE = "fanout_exchange";
       //队列名称
       static final String FANOUT_QUEUE_1 = "fanout_queue_1";
       //队列名称
       static final String FANOUT_QUEUE_2 = "fanout_queue_2";
    
       public static void main(String[] args) throws Exception { 
    
           //创建连接
           Connection connection = ConnectionUtil.getConnection();
    
           // 创建频道
           Channel channel = connection.createChannel();
    
           /** * 声明交换机 * 参数1：交换机名称 * 参数2：交换机类型，fanout、topic、direct、headers */
           channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
    
           // 声明（创建）队列
           /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */
           channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
           channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);
    
           //队列绑定交换机
           channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");
           channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");
    
           for (int i = 1; i <= 10; i++) { 
               // 发送信息
               String message = "你好；小兔子！发布订阅模式--" + i;
               /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */
               channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
               System.out.println("已发送消息：" + message);
           }
    
            // 关闭资源
            channel.close();
            connection.close();
        }
    }

3.消息消费者1：

import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * 消息消费者1 */
    public class Consumer1 { 
    
       public static void main(String[] args) throws Exception { 
           Connection connection = ConnectionUtil.getConnection();
    
           // 创建频道
           Channel channel = connection.createChannel();
    
           //声明交换机
           channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
    
           // 声明（创建）队列
           /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */
           channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);
    
           //队列绑定交换机
           channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, "");
    
            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){ 
                @Override
                /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
                }
            };
            //监听消息
            /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */
            channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer);
        }
    }

4.消息消费者2：

import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * 消息消费者2 */
    public class Consumer2 { 
    
       public static void main(String[] args) throws Exception { 
           Connection connection = ConnectionUtil.getConnection();
    
           // 创建频道
           Channel channel = connection.createChannel();
    
           //声明交换机
           channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
    
           // 声明（创建）队列
           /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */
           channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);
    
           //队列绑定交换机
           channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHAGE, "");
    
           //创建消费者；并设置消息处理
           DefaultConsumer consumer = new DefaultConsumer(channel){ 
               @Override
               /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                   //路由key
                   System.out.println("路由key为：" + envelope.getRoutingKey());
                   //交换机
                   System.out.println("交换机为：" + envelope.getExchange());
                   //消息id
                   System.out.println("消息id为：" + envelope.getDeliveryTag());
                   //收到的消息
                   System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));
               }
           };
           //监听消息
           /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */
           channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer);
       }
    }

## 三。程序运行效果 ##

1.启动消息生产者，发送10条消息至交换机fanout\_exchange中：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 1]  
2.查找到fanout\_exchange交换机绑定队列fanout\_queue\_1、fanout\_queue\_2，将消息保存至队列中：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 2]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 3]  
3.启动消息消费者1，声明交换机fanout\_exchange，并绑定fanout\_queue\_1队列，消费其中的消息：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 4]  
4.启动消息消费者2，声明交换机fanout\_exchange，并绑定fanout\_queue\_2队列，消费其中的消息：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 5]  
**结论**:

**交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。**

**发布订阅模式与工作队列模式的区别**

1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。

2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。

3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机 。

## 四。模式总结 ##

**1、简单模式 HelloWorld**

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

**2、工作队列模式 Work Queue**

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

**3、发布订阅模式 Publish/subscribe**

需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

**4、路由模式 Routing**

需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

**5、通配符模式 Topic**

需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70]: /images/20221120/129d5784b13c4d5cb901cc1c6fc6c576.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 1]: /images/20221120/ca558ba4b6854af5b501af23658a64e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 2]: /images/20221120/8fe413fadd664189adbba5488d45cfc5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 3]: /images/20221120/8e290d719dcf413eaa94a92bd2de1b30.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 4]: /images/20221120/94de6e733e934e37a93e59847dbe4cb0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70 5]: /images/20221120/627672cc0a2f45ddbae4bc552cfd3b42.png