RabbitMQ 订阅发布

今天来学习下 RabbitMQ 的消息订阅发布模式：pub/sub。

概念

发布订阅模式：一个生产者发布消息，多个消费者订阅该生产者，这样同一条消息可以被多个消费者消费。

模型

解读

一个生产者，多个消费者。

每个消费者都有自己的队列。

每个队列都绑定到交换器上。

生产者不是直接把消息发送到队列，而是发到了图中的“X”，也就是交换器（Exchange），交换器根据相应的路由绑定到对应的队列。

常见应用场景：用户注册后，短信、邮箱都通知用户注册成功。

代码示例

Send

package com.yakai.rabbitmq.ps;

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_fanout";
    //交换器类型是分发
    private static final String EXCHANGE_TYPE = "fanout";

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

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

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

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

启动 Send，登录localhost:15672控制台查看Exchange 一栏，发现交换器已经声明：

疑问

那么此时的消息在哪里呢？答案是丢失，这是因为消息只发送到 Exchange，但是 Exchange 没有存储消息的功能，队列才是内存中消息存放的地方，而此时Exchange 没有与任何队列绑定，刚才发的那条消息自然就丢失了。

Recv1

让消费者声明队列，然后 Send 根据路由绑定规则，通过 Exchange，绑定到对应的队列。

package com.yakai.rabbitmq.ps;

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

import java.io.IOException;

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

        //绑定交换器
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");

        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.ps;

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

import java.io.IOException;

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

        //绑定交换器
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");

        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[2] :"+msg);

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

测试

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

Send:

Send: hello pub/sub

Recv1:

Recv[1] :hello pub/sub

Recv2:

Recv[2] :hello pub/sub

现象

Send 发送了一条消息，Recv1和 Recv2 都接收到了这条消息。实现了 RabbitMQ 的发布订阅机制。

总结

在[RabbitMQ 基础概念详解](link: http://zhengyk.cn/2018/03/20/rabbitmq/basic-concept/)中了解了 Exchange、 RoutingKey、BindingKey等相关概念。 Exchange 类型主要有 fanout、direct、topic、headers四种，我们正是使用了 fanout 这一类型的交换器实现了订阅发布。

fanout 特点

它会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中。当使用此类型时，RoutingKey 和 BindingKey 默认使用空字符串。