RabbitMQ（五）：路由模式-蒲公英云

一、路由模式

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

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

二、direct交换机

生产者申明一个direct类型交换机，然后发送消息到这个交换机指定路由键。
消费者指定消费这个交换机的这个路由键，即可接收到消息，其他消费者收不到。

上章节中的生产者：

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

第二个参数就是路由键

消费者:

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

第三个参数就是路由键

三、代码演示

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

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

连接RabbitMQ工具类

package cn.saytime.rabbitmq.util;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/** * RabbitMQ连接工具类 */
public class ConnectionUtil { 
    private static final String host = "192.168.239.128";
    private static final int port = 5672;
    /** * 获取RabbitMQ Connection连接 * @return * @throws IOException * @throws TimeoutException */
    public static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
// connectionFactory.setUsername("test");
// connectionFactory.setPassword("123456");
// connectionFactory.setVirtualHost("/vhost_test");
        return connectionFactory.newConnection();
    }
}

如上所示，如果配置有用户名密码以及vhost，则配置即可。

生产者

package cn.saytime.rabbitmq.routing;
import cn.saytime.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/** * 生产者 */
public class Send {
    private static final String EXCHANGE_NAME = "test_exchange_fanout";
    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 cn.saytime.rabbitmq.routing;
import cn.saytime.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
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交换机，或者先执行一次生产者（执行时会判断交换机是否存在，不存在则创建交换机），这样保证交换机存在，不然直接启动消费者会提示交换机不存在。

这里写图片描述

注意点

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

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

Send

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

Recv

[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");