RabbitMQ路由模式

墨蓝 2022-05-16 04:39 266阅读 0赞

### 路由模式 ###

一个生产者，发送消息  
每个消费者，都有一个独立的队列

消息发送到交换机，交换机发送到每个队列  
根据key，是否相等，来接收消息  
![这里写图片描述][70]

### Send ###

生产者

package cn.itcast.rabbitmq.routing;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            // 消息内容
            String message = "删除商品，id=1000";
            channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());
            System.out.println(" 后台系统 '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }

### Recv ###

消费者1

package cn.itcast.rabbitmq.routing;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv { 
    
        private final static String QUEUE_NAME = "test_queue_direct1";
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 前台系统： '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

### Recv2 ###

消费者2

package cn.itcast.rabbitmq.routing;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv2 { 
    
        private final static String QUEUE_NAME = "test_queue_direct2";
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 搜索系统： '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

### 启动发送者 ###

![这里写图片描述][70 1]

### 启动消费者 ###

此时，有五个绑定队列

消费者1，两个  
消费者2，三个  
![这里写图片描述][70 2]

### 测试 ###

发送消息，依次将key设置为  
insert、update、delete

消费者1  
前台系统，接收了2条消息  
![这里写图片描述][70 3]  
消费者2  
搜索系统，接收了3条消息  
![这里写图片描述][70 4]

[70]: /images/20220516/bf9d0625c9be42b882e25340fe48f0ae.png
[70 1]: /images/20220516/6788ef42ea4e465080e9fd771efd51d3.png
[70 2]: /images/20220516/bfccf9c38f18439d86fa3728345a4381.png
[70 3]: /images/20220516/aa9809eaab4043f69712d204043f86ed.png
[70 4]: /images/20220516/0dbdba76b10c47e5a3054a3d7b46b7c6.png