fanout广播模式

我会带着你远行 2022-06-11 06:39 168阅读 0赞

RabbitMQ完整的消息模型如下:

![这里写图片描述][2016-02-18_56c53cbd5232a.jpg]

生产者P会通过交换器X所设定的规则将消息随机的发送到各个队列上边。

交换器的规则有下面的几种：  
1.direct （直连）

直连交换机是一种带路由功能的交换机，一个队列会和一个交换机绑定，除此之外再绑定一个`routing_key`，当消息被发送的时候，需要指定一个`binding_key`，这个消息被送达交换机的时候，就会被这个交换机送到指定的队列里面去。同样的一个`binding_key`也是支持应用到多个队列中的。

这样当一个交换机绑定多个队列，就会被送到对应的队列去处理。

适用场景：有优先级的任务，根据任务的优先级把消息发送到对应的队列，这样可以指派更多的资源去处理高优先级的队列。

2.topic （主题）

主题交换机，发送到主题交换机上的消息需要携带指定规则的routing\_key，主题交换机会根据这个规则将数据发送到对应的(多个)队列上。  
3.headers （标题）

路由器和交换机路由的规则是通过Headers信息来交换的，这个有点像HTTP的Headers。将一个交换机声明成首部交换机，绑定一个队列的时候，定义一个Hash的数据结构，消息发送的时候，会携带一组hash数据结构的信息，当Hash的内容匹配上的时候，消息就会被写入队列。  
4.fanout （分发）

扇形交换机是最基本的交换机类型，它所能做的事情非常简单———广播消息。扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。因为广播不需要“思考”，所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

下面我们做一个通过路由实现消息传递的例子:

1.生产者

public class Producer {
            private static final String EXCHANGE_NAME = "test";
    
            public static void main(String[] argv) throws Exception {
    
                ConnectionFactory factory = new ConnectionFactory();
                factory.setHost("localhost");
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();
                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
                for(int i = 0 ; i < 5; i++){
                    String message = "Hello World! " + i;
                    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
                    System.out.println(" [x] Sent '" + message + "'");
                }
                channel.close();
                connection.close();
            }
    
    }

通过

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

设置了一个名称为test的具有广播功能的交换器

2.消费者

public class Consumer1 {
    
            private static final String EXCHANGE_NAME = "test";
    
            public static void main(String[] argv) throws Exception {
                ConnectionFactory factory = new ConnectionFactory();
                factory.setHost("localhost");
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();
    
                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
                String queueName = channel.queueDeclare().getQueue();
                channel.queueBind(queueName, EXCHANGE_NAME, "");
    
                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");
                        System.out.println(" [x] Received '" + message + "'");
                    }
                };
                channel.basicConsume(queueName, true, consumer);
            }
    
    }

3.消费者2

public class Consumer2 {
            private static final String EXCHANGE_NAME = "test";
            public static void main(String[] argv) throws Exception {
                ConnectionFactory factory = new ConnectionFactory();
                factory.setHost("localhost");
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();
                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
                String queueName = channel.queueDeclare().getQueue();
                channel.queueBind(queueName, EXCHANGE_NAME, "");
                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");
                        System.out.println(" [x] Received '" + message + "'");
                    }
                };
                channel.basicConsume(queueName, true, consumer);
            }
    }

通过

String queueName = channel.queueDeclare().getQueue();  
  
创建了一个可以自动删除的临时队列

参考文章:

http://www.jianshu.com/p/469f4608ce5d

[2016-02-18_56c53cbd5232a.jpg]: /images/20220611/f9488251274c432caf8293a6a5a1d528.png