【RabbitMq】fanout消息模型（广播）

你的名字 2022-10-28 03:58 140阅读 0赞

# Fanout消息模型 #

`fanout`扇出 也称广播

![fanout.jpg][]

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

*  可以有多个消费者
 *  每个消费者都有自己的`queue`队列（临时，消费完断开连接后自动删除）
 *  每个队列都要绑定到`exchange`交换机
 *  生产者发送的消息，只能发送到交换机，交换机决定来发送给哪个队列，生产者无法决定
 *  交换机把收到的消息发送给每一个队列，实现一条消息被多个消费者消费

## 生产者 ##

public class Provider { 
        @Test
        public void test() throws IOException { 
            Connection connection = RabbitMqUtil.getConnection();
            Channel channel = connection.createChannel();
            //通道绑定声明交换机
            channel.exchangeDeclare("logs","fanout");
            //生产消息
            channel.basicPublish("logs","",null,"fanout type message".getBytes());
            RabbitMqUtil.close(channel,connection);
        }
    }

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

## 多消费者 ##

**customer1：**

public class Customer1 { 
        public static void main(String[] args) throws IOException { 
            Connection connection = RabbitMqUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //通道绑定交换机
            channel.exchangeDeclare("logs","fanout");
            //创建临时队列
            String queue = channel.queueDeclare().getQueue();
            //将队列和交换机进行绑定
            channel.queueBind(queue,"logs","");
            //消费消息
            channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("fanout模型 消费者一消费:"+new String(body));
                }
            });
    
        }
    }

**customer2：**

public class Customer2 { 
        public static void main(String[] args) throws IOException { 
            Connection connection = RabbitMqUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //通道绑定交换机
            channel.exchangeDeclare("logs","fanout");
            //创建临时队列
            String queue = channel.queueDeclare().getQueue();
            //将队列和交换机进行绑定
            channel.queueBind(queue,"logs","");
            //消费消息
            channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("fanout模型 消费者二消费:"+new String(body));
                }
            });
    
        }
    }

**customer3：**

public class Customer3 { 
        public static void main(String[] args) throws IOException { 
            Connection connection = RabbitMqUtil.getConnection();
            Channel channel = connection.createChannel();
    
            //通道绑定交换机
            channel.exchangeDeclare("logs","fanout");
            //创建临时队列
            String queue = channel.queueDeclare().getQueue();
            //将队列和交换机进行绑定
            channel.queueBind(queue,"logs","");
            //消费消息
            channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("fanout模型 消费者三消费:"+new String(body));
                }
            });
    
        }
    }

![消费者1.jpg][1.jpg]

![消费者2.jpg][2.jpg]

![消费者3.jpg][3.jpg]

[fanout.jpg]: /images/20221024/0fa37d7ae0924d2e8aa9d48594bad378.png
[1.jpg]: /images/20221024/71ab05c8a94d4554b426ba694c887bb6.png
[2.jpg]: /images/20221024/3e14215a894f4898b6c8438f5a2b4fd8.png
[3.jpg]: /images/20221024/ac995400d9824aef97ed098d5d8f737c.png