RabbitMQ 学习---- Fanout 广播模型

Posted 2022-11-23 RAIN 7

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ 学习---- Fanout 广播模型相关的知识，希望对你有一定的参考价值。

文章目录

RabbitMQ 学习（五）---- Fanout 广播模型

RabbitMQ 学习（五）---- Fanout 广播模型

这种广播模式很像，就是生产者把通知的内容都录音到电台（exange）里，消费者如果想要听到，必须使用该频率的耳机（queue 绑定 exange）去听才能收到。所有消费者只要通过一个队列进行绑定，那么都能听到，都能消费该信息，这就是广播模式。

（1）消息发送流程

在广播模式下，消息发送流程是这样的：

可以有多个消费者
每个消费者有自己的queue（队列）
每个队列都要绑定到Exchange（交换机）
生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
交换机把消息发送给绑定过的所有队列
队列的消费者都能拿到消息。实现一条消息被多个消费者消费

（2）生产者

声明交换机（名字+类型）
生产者使用信道发送消息到交换机中

package fanout;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import utils.RabbitMQUtils;

import java.io.IOException;

public class FProvider 
    public static void main(String[] args) 
        Connection connection = RabbitMQUtils.getConnect();
        Channel channel = null;

        try 
            // 创建信道
            channel = connection.createChannel();

            // 声明交换机
            channel.exchangeDeclare("exch", "fanout");

            String body = "广播模式发送消息!";

            // 信道使用交换机接收发送的消息，routineKey在这种广播模式下用不到
            channel.basicPublish("exch", "", MessageProperties.PERSISTENT_TEXT_PLAIN, body.getBytes());

         catch (IOException e) 
            
            e.printStackTrace();
        finally 
            RabbitMQUtils.close(channel,connection);

（3）消费者

声明交换机
声明临时队列
临时队列监听交换机
消费者使用队列接收消息

消费者1

package fanout;

import com.rabbitmq.client.*;
import utils.RabbitMQUtils;

import java.io.IOException;

public class FCustomer1 
    public static void main(String[] args) 
        Connection connection = RabbitMQUtils.getConnect();
        try 
            Channel channel = connection.createChannel();

            // 声明交换机，与生产者一致
            channel.exchangeDeclare("exch", "fanout");

            // 声明临时队列
            String queue = channel.queueDeclare().getQueue();

            // 交换机与队列进行绑定，临时队列绑定交换机才能听到广播的内容，routingKey不起作用
            channel.queueBind(queue, "exch", "");

            // 使用临时队列接收 生产者在交换机中 广播的消息
            channel.basicConsume(queue, false, new DefaultConsumer(channel)
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 
                    System.out.println(new String(body));;
                
            );


         catch (IOException e) 
            e.printStackTrace();