RabbitMQ Routing 模式

上篇学习总结了 RabbitMQ 的 sub/pub 模式，今天接着学习它的 Routing 模式。此模式使用了“direct”类型的 Exchange。

模型

解读

队列一的 BindingKey 是“error”，队列二的BindingKey 是“info”、“error”和“warning”，Exchange 的类型是 direct，当 RoutingKey 是 error 时，Exchange 将消息发送到 队列一，当 RoutingKey 是 info 或 error 或 warning 时，Exchange 将消息发送到队列二。

代码示例

Send

package com.yakai.rabbitmq.direct;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.yakai.rabbitmq.util.ConnectionUtil;

public class Send {

    private static final String EXCHANGE_NAME = "test_ex_direct";
    //交换器类型是分发
    private static final String EXCHANGE_TYPE = "direct";

    public static void main(String[] args) throws Exception {

        //获取一个连接
        Connection connection = ConnectionUtil.getConnection();
        //从连接中获取一个通道
        Channel channel = connection.createChannel();

        //声明交换器
        channel.exchangeDeclare(EXCHANGE_NAME,EXCHANGE_TYPE);
        String msg = "error logs";
        //发送消息
        String routingKey = "error";
        channel.basicPublish(EXCHANGE_NAME,routingKey,null,msg.getBytes());
        System.out.println("Send: "+msg);

        //关闭通道、连接
        channel.close();
        connection.close();
    }
}

注：Send 写完后先运行一下，让 RabbitMQ 中先生成这个交换器，防止Recv 运行的时候报错

Recv1

package com.yakai.rabbitmq.direct;

import com.rabbitmq.client.*;
import com.yakai.rabbitmq.util.ConnectionUtil;

import java.io.IOException;

public class Recv1 {
    private static final String QUEUE_NAME = "test_routing_queue_1";
    private static final String EXCHANGE_NAME = "test_ex_direct";
    public static void main(String[] args) throws Exception{
        //获取一个连接
        Connection connection = ConnectionUtil.getConnection();
        //从连接中获取一个通道
        Channel channel = connection.createChannel();
        //声明一个队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //绑定交换器 Recv1接收routingKey=“error”的队列一发送的消息
        String routingKey = "error";
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,routingKey);

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body,"utf-8");
                System.out.println("Recv[1] :"+msg);

            }
        };
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

Recv2

package com.yakai.rabbitmq.direct;

import com.rabbitmq.client.*;
import com.yakai.rabbitmq.util.ConnectionUtil;

import java.io.IOException;

public class Recv2 {
    private static final String QUEUE_NAME = "test_routing_queue_2";
    private static final String EXCHANGE_NAME = "test_ex_direct";
    public static void main(String[] args) throws Exception{
        //获取一个连接
        Connection connection = ConnectionUtil.getConnection();
        //从连接中获取一个通道
        Channel channel = connection.createChannel();
        //声明一个队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //绑定交换器  routingKey 三个
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"error");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"info");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"warning");

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body,"utf-8");
                System.out.println("Recv[1] :"+msg);

            }
        };
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

注：channel.queueBind(String queue,String exchange,String routingKey);方法中的参数 routingKey 其实准确地来说应该是 bindingKey，其实在某些情形下，RoutingKey 和 BindingKey可以认为是一个东西，因为在 direct 交换器类型下，RoutingKey 和 BindingKey 需要完全匹配才可能使用，所以代码中的 API 使用了此种写法会显得很方便

测试

先启动 Recv1、Recv2，再启动 Send，观察数据：

Send:

Send: error logs

Recv1:

Recv[1] :error logs

Recv2:

Recv[2] :error logs

发送“error”信息，Recv1和 Recv2都可以接收到。

修改Send代码：

String msg = "info logs";
//发送消息
String routingKey = "info";

其他不变，继续测试，观察控制台打印：

Send:

Send: info logs

Recv1:

控制台无变化

Recv2:

Recv[2] :info logs

总结

“direct”类型的 Exchange 会把消息路由到 BindingKey 和 RoutingKey 完全匹配的队列中