RabbitMQ广播模式（动态生成queue）

妖狐艹你老母 2023-06-24 04:46 16阅读 0赞

RabbitMQ的广播机制和ActiveMQ有所不同。  
先来梳理下RabbitMQ中消息从产生到消费的流程吧：  
![在这里插入图片描述][20191220154408767.png]  
而exchange 存在多种类型，这里就只说广播模式（fanout）了。在广播模式中，一个exchange对应多个queue，会向每个queue都发送信息，然后不同的queue再由其对应的消费者消费信息，即完成了广播。  
因为广播模式中不关注routingkey和queue，只需要queue的queue name唯一即可，所以这里把routingkey移出来了，实际上还是会经过的哦。

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

### 1.新建一个spring boot 项目并引入官方的amqp包 ###

<dependency>
    	<groupId>org.springframework.amqp</groupId>
    	<artifactId>spring-rabbit</artifactId>
    </dependency>

### 2.添加RabbitMQ连接参数 ###

spring.rabbitmq.host=xx
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=xx
    spring.rabbitmq.password=xx

### 3.创建生产者 ###

同时添加一个配置参数`rabbit.exchange`，用来动态指定exchange（比如使用apollo或者spring cloud config）  
增加配置参数

rabbit.exchange=testExchange

增加生产者

@Component
    public class Producer { 
        @Autowired
        private RabbitTemplate rabbitTemplate;
        @Value("${rabbit.exchange}")
        private String exchange;
    
        public void sendInfo(){ 
            Message message = new Message("123".getBytes(),new MessageProperties());
            rabbitTemplate.send(exchange,"",message);
        }
    }

### 4.动态创建queue和消费者 ###

因为需要执行`createQueue`方法才能生成一个queue和消费者，所以这里先用`@Component`指定扫描当前类，再用`@PostConstruct`指定扫描时执行该方法。  
因为创建的queue是临时queue，当消费者消失时，该queue就会自动删除，因为创建queue也是由rabbitMQ自行生成的，所以queue name一定是唯一的。这样在集群部署时，就可以做到即开即用了，就算关闭了服务，对应的queue也会自动消失。

@Component
    public class Consumer { 
        @Autowired
        private RabbitTemplate rabbitTemplate;
        @Value("${rabbit.exchange}")
        private String exchange;
    
        @PostConstruct
        public void createQueue(){ 
            Channel channel = rabbitTemplate.getConnectionFactory().createConnection().createChannel(true);
    
            try{ 
                /** * 与生产者使用同一个交换机 */
                channel.exchangeDeclare(exchange, "fanout",true);
                /** * 获取一个随机的队列名称，使用默认方式，产生的队列为临时队列，在没有消费者时将会自动删除 */
                String queueName = channel.queueDeclare().getQueue();
    
                /** * 关联 exchange 和 queue ，因为是广播无需指定routekey，routingKey设置为空字符串 */
                // channel.queueBind(queue, exchange, routingKey)
                channel.queueBind(queueName, exchange, "");
    
                com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) { 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        String message = new String(body, "UTF-8");
    
                        if(StringUtils.isEmpty(message)){ 
                            return;
                        }
    
                        /** * 对信息做操作 */
                    }
    
                };
                //true 自动回复ack
                channel.basicConsume(queueName, true, consumer);
            }catch (Exception ex){ 
            }
        }
    }

### 5.手动在RabbitMQ控制界面创建exchange ###

如下图选择广播模式，并且为永久exchange，如果要设置临时exchange的话要修改4中的如下语句最后一个参数为`false`

channel.exchangeDeclare(exchange, "fanout",true);

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 1]

### 6.启动项目，测试 ###

在`Consumer`中打个断点，就能看到在启动时调用`createQueue`后所产生的队列名  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 2]  
到RabbitMQ控制台上搜索下，确实存在  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 3]  
点开看下对应的exchange  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 4]  
完全OK，发送条信息试下，也能够接收到  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 6]  
可以试下再重复第4步，再建一个queue，或者再起一个项目，你会发现两个消费者都能接收到信息。

然后我们把项目关闭，看看queue还在不在，明显已经木有了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 7]

[20191220154408767.png]: https://img-blog.csdnimg.cn/20191220154408767.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191220154210933.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191220174005287.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191220175127917.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20191220175140969.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20191220175159451.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20191220175356566.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20191220175407625.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/2019122017562153.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lleWluZ2xpbmdmZW5n,size_16,color_FFFFFF,t_70