RabbitMQ 路由模式

系统管理员 2024-03-25 20:17 31阅读 0赞

##### 一、路由模式 #####

官方内容参考：[http://www.rabbitmq.com/tutorials/tutorial-four-java.html][http_www.rabbitmq.com_tutorials_tutorial-four-java.html]

跟订阅模式类似，只不过在订阅模式的基础上加上了类型，订阅模式是分发到所有绑定到交换机的队列，

路由模式只分发到绑定在交换机上面指定路由键的队列。

![d5134460a8be42d2ac1a1ea908155388.png][]

##### 二、direct交换机 #####

生产者申明一个direct类型交换机，然后发送消息到这个交换机指定路由键。

消费者指定消费这个交换机的这个路由键，即可接收到消息，其他消费者收不到。

上章节中的

生产者：第二个参数是路由键

channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());

消费者:第三个参数是路由键

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

##### 三、代码演示 #####

同样的，只是交换机类型改为direct，加了个路由键而已。

这里按图演示两个，即表示一个日志事件，根据日志类型进行处理。

###### 生产者 ######

package com.rabbitmq.routing;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 生产者
     */
    public class Send {
    
        private static final String EXCHANGE_NAME = "test_exchange_routing";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtil.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 声明一个direct路由交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    
    
            // 发送info路由键消息
            String infoMessage = "hello, info";
            channel.basicPublish(EXCHANGE_NAME, "info", null, infoMessage.getBytes());
            System.out.println(" [x] Sent routing info message : '" + infoMessage + "'");
    
            // 发送error路由键消息
            String errorMessage = "hello, error";
            channel.basicPublish(EXCHANGE_NAME, "error", null, errorMessage.getBytes());
            System.out.println(" [x] Sent routing error message :  '" + errorMessage + "'");
    
            channel.close();
            connection.close();
    
        }
    }

###### info日志消费者 ######

package cn.saytime.rabbitmq.routing;
    
    import cn.saytime.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * info日志消费者
     */
    public class Recv {
    
        // info日志队列
        private static final String QUEUE_NAME = "test_queue_routing_info";
        private static final String EXCHANGE_NAME = "test_exchange_routing";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 打开通道
            Channel channel = connection.createChannel();
    
            // 申明要消费的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
    
            // 这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。
            channel.basicQos(1);
    
            // 创建一个回调的消费者处理类
            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);
                    System.out.println(" [1] Received '" + message + "'");
    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        System.out.println(" [1] done ");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
            // 消费消息
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
        }
    
    }

###### error日志消费者 ######

package com.rabbitmq.routing;
    
    import com.rabbitmq.client.*;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * error日志消费者
     */
    public class Recv2 {
    
        // error日志队列
        private static final String QUEUE_NAME = "test_queue_routing_error";
        private static final String EXCHANGE_NAME = "test_exchange_routing";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 打开通道
            Channel channel = connection.createChannel();
    
            // 申明要消费的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
    
            // 这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。
            channel.basicQos(1);
    
            // 创建一个回调的消费者处理类
            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);
                    System.out.println(" [2] Received '" + message + "'");
    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        System.out.println(" [2] done ");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
            // 消费消息
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    
    }

##### 四、测试结果 #####

提前在管理控制台创建一个direct交换机，或者先执行一次生产者（执行时会判断交换机是否存在，不存在则创建交换机），这样保证交换机存在，不然直接启动消费者会提示交换机不存在。

![d8a233bf312c4eb9bdeeacd45e1902d0.png][]

###### 注意点 ######

如果在没有队列绑定在交换机上面时，往交换机发送消息会丢失，之后绑定在交换机上面的队列接
    收不到之前的消息，也就是先执行第一次发送，创建了交换机，但是还没有队列绑定在交换机上面，
    如果这次发送的消息就会丢失。

然后启动两个消费者，再执行生产者。

Send

[x] Sent routing info message : 'hello, info'
     [x] Sent routing error message :  'hello, error'

Recv1

[1] Received 'hello, info'
     [1] done

Recv2

[2] Received 'hello, error'
     [2] done

我们可以看到生产者往info路由键发送消息时，只有执行消费info路由键的消费者才能接收到消息，
    error路由键同样。

##### 五、多绑定情况 #####

###### 1. 同一个消费者绑定队列多个路由键 ######

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "warn");

如果一个消费者绑定了这三个路由键，那么当生产者发送其中一个路由键消息时，该消费者都能接收到消息。

###### 2. 多个消费者绑定相同路由键 ######

即消费者1绑定info，消费者2 绑定info, error

那么生产者发送info路由键消息时，消费者1, 2都能接收到消息，发送error路由键消息时，只有消费者2能接收到消息。

[http_www.rabbitmq.com_tutorials_tutorial-four-java.html]: http://www.rabbitmq.com/tutorials/tutorial-four-java.html
[d5134460a8be42d2ac1a1ea908155388.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/ddb6922cf26b4f68a26c81e3f0904993.png
[d8a233bf312c4eb9bdeeacd45e1902d0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/295852eaac70480faadc8d5eefe722f5.png