《RabbitMQ系列教程-第四章-03-RabbitMQ工作模式之Pub/Sub发布订阅模式》

约定不等于承诺〃 2023-01-02 03:24 4阅读 0赞

# 教程说明 #

*  本系列教程目录大纲：[《RabbitMQ系列教程-目录大纲》][RabbitMQ_-]

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

*  本系列教程配套代码：[https://gitee.com/Horizon1024/rabbitmt.git][https_gitee.com_Horizon1024_rabbitmt.git]（码云地址）

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

# RabbitMQ工作模式之Pub/Sub发布订阅模式 #

### 4.3.1 简介 ###

在发布订阅模式中，`Producer`发送消息到指定的交换机（`Exchange`）中，由`Exchange`绑定不同的`Queues`，消费者依旧监听这些队列进行消费（`work`模式使用的是默认的交换机，空字符串`""`）

![在这里插入图片描述][20210104161821880.png_pic_center]

> tips：**交换机（Exchange）只负责将消息转发到绑定的队列（Queues）中，不会存储消息，如果Exchange没有绑定Queues或者Exchange不能将消息路由到指定的Queues中时，此消息将会丢失**；

一个Exchange可以有`direct`、`topic`、`headers`、`fanout`四种类型，不同类型的Exchange具备不同的消息路由功能，**`Pub/Sub`模式则重点强调的是`fanout`类型的`Exchange`交换模式，`Pub/Sub`模式也叫分列模式；**

> Pub/Sub模式官网介绍：https://www.rabbitmq.com/tutorials/tutorial-four-java.html

### 4.3.2 生产者 ###

package com.lscl.rabbitmq;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class Producer03_PubSub { 
    
        public static void main(String[] args) throws Exception { 
            // 创建连接工厂,用于获取频道channel
            ConnectionFactory factory = new ConnectionFactory();
    
            factory.setHost("192.168.40.132");
            factory.setPort(5672);
            factory.setUsername("lscl");
            factory.setPassword("admin");
            factory.setVirtualHost("/lscl");
    
            // 2.创建连接
            Connection connection = factory.newConnection();
    
            // 3.创建频道
            Channel channel = connection.createChannel();
    
            String exchangeName = "test_fanout";
    
            //5. 创建交换机
            /* exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) 1. exchange:交换机名称 2. type:交换机类型 DIRECT("direct"),：定向 FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。 TOPIC("topic"),通配符的方式 HEADERS("headers");参数匹配 3. durable:是否持久化 4. autoDelete:自动删除 5. internal：内部使用。 一般false 6. arguments：参数 */
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
    
            //6. 创建队列
            String queue1Name = "test_fanout_queue1";
            String queue2Name = "test_fanout_queue2";
    
            channel.queueDeclare(queue1Name,true,false,false,null);
            channel.queueDeclare(queue2Name,true,false,false,null);
            //7. 绑定队列和交换机
            /* queueBind(String queue, String exchange, String routingKey) 参数： 1. queue：队列名称 2. exchange：交换机名称 3. routingKey：路由键，绑定规则 如果交换机的类型为fanout ，routingKey设置为"" */
            channel.queueBind(queue1Name,exchangeName,"");
            channel.queueBind(queue2Name,exchangeName,"");
    
            String body = "呼叫各位";
            //8. 发送消息
            channel.basicPublish(exchangeName,"",null,body.getBytes());
    
            //9. 释放资源
            channel.close();
            connection.close();
        }
    }

运行完毕之后，查看RabbitMQ UI面板，发现多了一个Exchange（test\_fanout）和两个Queues（`test_fanout_queue1`、`test_fanout_queue2`）

### 4.3.3 消费者1 ###

package com.lscl.rabbitmq;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer03_PubSub { 
    
        public static void main(String[] args) throws Exception { 
            // 创建连接工厂,用于获取频道channel
            ConnectionFactory factory = new ConnectionFactory();
    
            factory.setHost("192.168.40.132");
            factory.setPort(5672);
            factory.setUsername("lscl");
            factory.setPassword("admin");
            factory.setVirtualHost("/lscl");
    
            // 2.创建连接
            Connection connection = factory.newConnection();
    
            // 3.创建频道
            Channel channel = connection.createChannel();
    
            String queueName = "test_fanout_queue1";
            channel.queueDeclare(queueName, true, false, false, null);
    
            // 接收消息
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("consumer-1 body：" + new String(body));
                }
            };
            channel.basicConsume(queueName, true, consumer);
            
            // 不释放资源,让rabbitmq一直监听
        }
    }

### 4.3.4 消费者2 ###

package com.lscl.rabbitmq;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer04_PubSub { 
    
        public static void main(String[] args) throws Exception { 
            // 创建连接工厂,用于获取频道channel
            ConnectionFactory factory = new ConnectionFactory();
    
            factory.setHost("192.168.40.130");
            factory.setPort(5672);
            factory.setUsername("lscl");
            factory.setPassword("admin");
            factory.setVirtualHost("/lscl");
    
            // 2.创建连接
            Connection connection = factory.newConnection();
    
            // 3.创建频道
            Channel channel = connection.createChannel();
    
            String queueName = "test_fanout_queue2";
            channel.queueDeclare(queueName, true, false, false, null);
    
            // 4.接收消息
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("consumer-2 body：" + new String(body));
                }
            };
            channel.basicConsume(queueName, true, consumer);
            
            // 不释放资源,让rabbitmq一直监听
        }
    }

### 4.3.5 Pub/Sub模式小结 ###

在`Pub/Sub`模式下（`Exchange`类型为`Fanout`，也叫分列模式），消息首先被发送到`Exchange`，由`Exchange`路由到绑定的`Queue`中，类似于我们微信群，有什么事在群里面发送一下，群里面的人都能看得到；这样就不需要每个人单独发送消息了；

> **需要注意的两点：**  
>   
>   
> **1、work、simple也会有交换机，他们使用的是默认的交换机**  
>   
>   
> **2、Exchange还可以绑定到另一个Exchange上**

`Pub/Sub`模式重点强调的是`Fanout`类型的`Exchange`交换模式，很多人也把`Pub/Sub`模式称为`分列模式`或者`Fanout`模式；

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

下一篇：[《RabbitMQ系列教程-第四章-04-RabbitMQ工作模式之Routing路由模式》][RabbitMQ_-_-04-RabbitMQ_Routing]

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

[RabbitMQ_-]: https://blog.csdn.net/Bb15070047748/article/details/112194912
[https_gitee.com_Horizon1024_rabbitmt.git]: https://gitee.com/Horizon1024/rabbitmt.git
[20210104161821880.png_pic_center]: /images/20221120/ece54732fe4e45fe89ccea745c6a7b7e.png
[RabbitMQ_-_-04-RabbitMQ_Routing]: https://blog.csdn.net/Bb15070047748/article/details/112184411