【RabbitMQ消息中间件】8.路由模式

上一篇讲解了RabbitMQ的“订阅模式”，本篇来讲解RabbitMQ的队列的“路由模式”。

其实严格来讲，RabbitMQ只有三种模式：“简单模式”、“work模式”以及“交换机模式”。

对于交换机模式来说，又分三种：“订阅模式”、“路由模式”、“通配符模式”，而他们之间的不同就是交换机类型的不同。

目前在实际开发中我们可能会遇到这种问题，生产者发布到消息队列的信息，消费者不一定全部需要，但是使用订阅模式的话，所有消费者都能够拿到，不满足只拿到自己所需信息的需求。

对于以上的需求，“路由模式”就可以进行实现。

路由模式的图示如下：

还是一个生产者、一个交换机，多个绑定交换机的队列，和多个连接不同队列的消费者。与之前的订阅模式不同的是，这里指定了交换机的类型是“direct”；而消费者将队列绑定到交换机时，指定了一个路由Key(error/info/warning等类型)。

如果生产者发送的消息中的路由key是“error”，那么消费者c1和c2都可以接收到，而如果生产者发送的消息中的路由kay是“info”或者“warning”，则只有消费者c2能够收到。

由此可见。“路由模式”的作用就是可以让消费者有选择性的接收消息。

“路由模式”的实现代码和之前的“订阅模式”的生产者和消费者代码差不多，对于生产者：

package cn.jack.rabbitmq.routing;
import cn.jack.rabbitmq.connection.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[] args) throws Exception {
        //获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        //从连接中创建通道
        Channel channel = connection.createChannel();
        
        //声明Exchange，指定交换机的类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        
        //消息内容1 info类型
        String message = "Hello World!";
        channel.basicPublish(EXCHANGE_NAME, "info",null, message.getBytes());
        System.out.println("[product] Send '"+ message +"'");
        
        //消息内容2 error类型
        message = "Has Error!";
        channel.basicPublish(EXCHANGE_NAME, "error",null, message.getBytes());
        System.out.println("[product] Send '"+ message +"'");
        
        //消息内容3 warning类型
        message = "Has Warning!";
        channel.basicPublish(EXCHANGE_NAME, "warning",null, message.getBytes());
        System.out.println("[product] Send '"+ message +"'");
        
        //关闭通道和连接
        channel.close();
        connection.close();
    }
}

在上面的代码中，首先生产者绑定交换机的时候，指定类型为“direct”，然后发布消息时指定不同消息的路由key。

然后消费者1的代码：

package cn.jack.rabbitmq.routing;
import cn.jack.rabbitmq.connection.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

public class Recv1 {

    private final static String QUEUE_NAME = "test_queue_direct_1";//队列名称
    
    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);
        
        //绑定队列到交换机，并指定了两个路由key
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");

        // 同一时刻服务器只会发一条消息给消费者
        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(" [consumer1] Received '" + message + "'");
            //休眠10ms
            Thread.sleep(10);
            // 返回确认状态
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

对于消费者1，在进行绑定队列到交换机时，设置了两个路由key，也就是它只需要接受路由key为“info”和“error”的消息。

然后消费者2的代码：

package cn.jack.rabbitmq.routing;
import cn.jack.rabbitmq.connection.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_direct_2";//队列名称
    
    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);
        
        //绑定队列到交换机，并指定了一个路由key
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "warning");

        // 同一时刻服务器只会发一条消息给消费者
        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(" [consumer2] Received '" + message + "'");
            //休眠10ms
            Thread.sleep(10);
            // 返回确认状态
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

对于消费者2，在进行绑定队列到交换机时，设置了一个路由key，也就是它只需要接受路由key为“warning”的消息。

下面首先运行两个消费者，然后运行生产者，他们的发送与接收情况如下：

可以看到，生产者一共发送了三种路由key的消息，而消费者按照自己绑定队列时的设置接收到需要的路由key的信息。

​