RabbitMQ——发布订阅模式

阅读完需：约 6 分钟

上文的工作队列模式是直接在生产者与消费者里声明好一个队列，这种情况下消息只会对应同类型的消费者。

显然这种只处理同种类型的消息是有弊端的。

举个用户注册的列子

门户网站，用户在注册完后一般都会发送消息通知用户注册成功（失败）。

如果在一个系统中，用户注册信息有邮箱、手机号，那么在注册完后会向邮箱和手机号都发送注册完成信息（假设都发送）。

利用 MQ 实现业务异步处理，如果是用工作队列的话，就会声明一个注册信息队列。注册完成之后生产者会向队列提交一条注册数据，消费者取出数据同时向邮箱以及手机号发送两条消息。但是实际上邮箱和手机号信息发送实际上是不同的业务逻辑，不应该放在一块处理。

这个时候就可以利用发布/订阅模式将消息发送到转换机（EXCHANGE），声明两个不同的队列（邮箱、手机），并绑定到交换机。这样生产者只需要发布一次消息，两个队列都会接收到消息发给对应的消费者，大致如下图所示。

在应用中，只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。

文章目录

• 1. 什么是发布订阅模式
• 2. 代码部分
  • 2.1 生产者
  • 2.2 邮件消费者
  • 2.3 短信消费者
  • 2.4 运行截图
• 3. 总结

1. 什么是发布订阅模式

简单解释就是，可以将消息发送给不同类型的消费者。做到发布一次，消费多个。下图取自于官方网站（RabbitMQ）的发布/订阅模式的图例：

P 表示为生产者、 X 表示交换机、C1C2 表示为消费者，红色表示队列。

2. 代码部分

2.1 生产者

public class ProducerFanout {

    private static final String EXCHANGE_NAME = "fanout_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        /** 1.创建新的连接 */
        Connection connection = MQConnectionUtils.newConnection();
        /** 2.创建通道 */
        Channel channel = connection.createChannel();
        /** 3.绑定的交换机 参数1交互机名称 参数2 exchange类型 */
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        /** 4.发送消息 */
        for (int i = 0; i < 10; i++)
        {
            String message = "用户注册消息：" + i;
            System.out.println("[send]：" + message);
            //发送消息
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("utf-8"));
            try {
                Thread.sleep(5 * i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        /** 5.关闭通道、连接 */
        channel.close();
        connection.close();
        /** 注意：如果消费没有绑定交换机和队列，则消息会丢失 */
    }

}

代码补充，channel.basicPublish(EXCHANGE_NAME, “”, null, message.getBytes(“utf-8”)); 其中第二个参数为空类似于表示全局广播，只要绑定到该队列上的消费者理论上是都可以收到的。

2.2 邮件消费者

public class ConsumerEmailFanout {

    private static final String QUEUE_NAME = "consumerFanout_email";
    private static final String EXCHANGE_NAME = "fanout_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        System.out.println("邮件消费者启动");
        /* 1.创建新的连接 */
        Connection connection = MQConnectionUtils.newConnection();
        /* 2.创建通道 */
        Channel channel = connection.createChannel();
        /* 3.消费者关联队列 */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        /* 4.消费者绑定交换机 参数1 队列 参数2交换机 参数3 routingKey */
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        DefaultConsumer 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("消费者获取生产者消息:" + msg);
            }
        };
        /* 5.消费者监听队列消息 */
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }

}

代码补充， channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, “”); 中第三个参数置为空时，可以接收到生产者所有的消息（生产者 routingKey 参数为空时）。

2.3 短信消费者

public class ConsumerSMSFanout {

    private static final String QUEUE_NAME = "ConsumerFanout_sms";
    private static final String EXCHANGE_NAME = "fanout_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        System.out.println("短信消费者启动");
        /* 1.创建新的连接 */
        Connection connection = MQConnectionUtils.newConnection();
        /* 2.创建通道 */
        Channel channel = connection.createChannel();
        /* 3.消费者关联队列 */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        /* 4.消费者绑定交换机 参数1 队列 参数2交换机 参数3 routingKey */
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        DefaultConsumer 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("消费者获取生产者消息:" + msg);
            }
        };
        /* 5.消费者监听队列消息 */
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }

}

代码补充， channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, “”); 中第三个参数置为空时，可以接收到生产者所有的消息（生产者 routingKey 参数为空时）

2.4 运行截图

先运行两个消费者，再运行生产者。如果没有提前将队列绑定到交换机，那么直接运行生产者的话，消息是不会发到任何队列里的。

生产者

短信消费者

邮件消费者

3. 总结

首先相对于工作模式，发布订阅模式引入了交换机的概念，相对其类型上更加灵活广泛一些。通过上文我们可以总结如下：

1.生产者不是直接操作队列，而是将数据发送给交换机，由交换机将数据发送给与之绑定的队列。从不加特定参数的运行结果中可以看到，两种类型的消费者（email，sms）都收到相同数量的消息。

1. 必须声明交换机，并且设置模式：channel.exchangeDeclare(EXCHANGE_NAME, “fanout”)，其中 fanout 指分发模式（将每一条消息都发送到与交换机绑定的队列）。
2. 队列必须绑定交换机：channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, “”);

生产者发送消息到交换机，多个消费者声明多个队列，与交换机进行绑定，队列中的消息可以被所有消费者消费，类似于QQ群消息

案例代码：https://www.lanzous.com/i5ydu6d