手把手一起入门RabbitMQ的六大使用模式（Java客户端）

普通队列、工作模式、发布订阅、路由模式、主题模式，RPC 模式。原理和代码，一个都不落下，一起入门 RabbitMQ 吧！

原文地址：手把手一起入门 RabbitMQ 的六大使用模式（Java 客户端）

为什么使用 MQ？

在这里我就不多说了，无非就是削峰、解耦和异步。这里没有很多关于 MQ 的理论和概念，只想手把手带你一起学习 RabbitMQ 的六大使用模式！

一、普通队列

我们发送消息和接收消息时，只需要直接指定队列的名字即可。这是最简单的一种使用场景。

生产者：使用 channel 发送消息时，直接指定 queueName。

public class Send {

    private static final String queueName = "hyf.hello.queue";

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

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        try (Connection cOnnection= factory.newConnection();
             Channel channel = connection.createChannel()){

            // 是否持久化（默认保存在内存，可以持久化到磁盘）
            boolean durable = false;
            // 是否独有（此 Connection 独有，通过其他 Connection 创建的 channel 无法访问此队列）
            boolean exclusive = false;
            // 是否自动删除队列（队列没有消费者时，删除）
            boolean autoDelete = false;
            channel.queueDeclare(queueName, durable, exclusive, autoDelete, null);

            String message = "Hello world3!";
            // 第一个参数是交换器名字，第二个参数是 routingKey（不使用交换器时，为队列名称），第三个参数是消息属性（AMQP.BasicProperties），第四个参数是消息
            channel.basicPublish("", queueName, null, message.getBytes());
            System.out.println("发布成功");
        }
    }
}

注意：使用 try-with-resources ，在程序结束时，我们不用显式调用 close() 方法来关闭资源。

消费者：也是用 channel 指定 queueName，然后绑定一个交付回调。

public class Receive {

    private static final String queueName = "hyf.hello.queue";

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

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        Connection cOnnection= factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(queueName, false, false, false, null);
        // 回调（接收 RabbitMQ 服务器发送过来的消息）
        DeliverCallback deliverCallback =  (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(message);
        };

        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

注意：这里我们可以不用 try-with-resource，因为消费者需要一直运行着。

关于普通队列，大家可以理解为下图：
在这里插入图片描述

二、工作模式(work queues)

普通队列中，都是一个消费者去消费队列，而在 work 模式中，是多个消费者同时去消费同一个队列。

生产者和消费者我们还是可以用回上面的代码。

1、循环轮询

默认情况下，RabbitMQ 将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。

这样会导致一个问题，即使其中一个消费者消费速度很快，已经消费完 RabbitMQ 消息，并且队列中还有未消费消息（已经分派给其他消费者），那么他也将在白白等待，RabbitMQ 而不会说将分派的消息回收重新分派给空闲的消费者。

2、自动提交消息 ack

默认情况下，消费者会不定时自动提交 ack，不管消息是否消费成功，而当 RabbitMQ 接收到消费者的 ack 消息后，会将消息添加删除标识来标识消息已被消费成功。但是这个自动 ack 机制会导致消息丢失和消息重复消费问题。

客户端还没消费某条消息，就自动提交了 ack，如果此时客户端宕机了，那么会导致这条消息消费失败；而 RabbitMQ 在接收到 ack 时，也将这条消息标记为已消费，那么也无法重新消费了。
客户端已经消费某条消息，但是还没自动提交 ack 就宕机了，此时就会导致消息重复消费，因为 RabbitMQ 没收到 ack 消息，那么这条消息没有被设置为删除标识，所以消费者还可以消费此条消息。

3、手动 ack 解决空闲消费者、消息丢失、消息重复消费

消费者：

a. 限制每次读取消息数量：

我们利用 basicQos() 方法来设置 prefetchCount(预期计数) 为1，即限制客户端每次都只读取一个消息，只有当这个消息消费完了，才能继续读取下一个消息。

b. 手动 ack：

接着我们需要关闭自动提交 ack，并且在消费完消息后，手动提交 ack。只有当 RabbitMQ 收到 ack 消息后，才会认定这个消息已经消费完了，继续给消费者推送下一条新消息。

最后看看代码：

public class Receive1 {

    private static final String queueName = "hyf.work.queue";

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

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        Connection cOnnection= factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(queueName, false, false, false, null);
        // 每次只读取一条消息
        channel.basicQos(1);
        DeliverCallback deliverCallback =  (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            ThreadUtil.sleep(2, TimeUnit.SECONDS);
            System.out.println(message);
            // 是否批量提交
            boolean multiple = false;
            // 手动 ack
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),multiple);
        };
        // 取消自动 ack
        boolean autoAck = false;
        channel.basicConsume(queueName, autoAck, deliverCallback, consumerTag -> {});
    }
}

总结：只有当我们使用了手动ack 和 prefetchCount = 1 ，工作模式才算成功启动。
在这里插入图片描述

4、扩展点：如何保证消息不丢失

当发送者发送消息到 RabbitMQ 后，RabbitMQ 会将消息缓存在内存中，而如果此时 RabbitMQ 宕机了，默认情况下，内存中的 queue 和 message 都会全部丢失。

而如果我们需要保证消息不丢失，那么需要告诉 RabbitMQ 如何做；此时我们需要做的是：将 queue 和 message 都设置为持久化。

queue 持久化：

private static final String queueName = "hyf.work.queue";

boolean durable = true;
channel.queueDeclare(queueName, durable, false, false, null);

注意：如果一开始 queue 已经定义为不持久化，那么我们不能重定义为持久化；当 RabbitMQ 检测到 queue 被重定义了，那么会返回一个错误来提示我们。

message 持久化：

private static final String queueName = "hyf.work.queue";

channel.basicPublish("", queueName,
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());

三、发布订阅模式(Publish/Subscribe)

上面的 work queue，每一个消息只能被一个消费者消费。而有些场景，我们需要一个消息可以被多个消费者消费；例如：用户下了订单，短信通知模块需要给用户发送一个短信通知，库存模块需要根据用户下单信息减去商品的库存等等，此时我们需要使用发布订阅模式。

1、交换器 exchange

要做发布订阅模式，我们首先需要使用到交换器，生产者不再直接利用 channel
往 queue 发送消息，而是将消息发送到交换器，让交换器来决定发送到哪些 queue 中。

RabbitMQ 提供了几个类型的交换器：direct、topic、headers 和 fanout。

使用发布订阅模式，我们只需要使用 fanout 类型的交换器，fanout 类型的交换器，会将消息发送到所有绑定到此交换器的 queue。

2、生产者发送消息：

利用 channel 声明交换器：

// 声明交换器名字和类型
channel.exchangeDeclare(exchangeName,"fanout");

接着我们就可以直接指定交换器进行消息发布：

// 第二个参数是 queueName/routingKey
channel.basicPublish(exchangeName , "", null, message.getBytes())

完整代码：

public class Send {

    private static final String exchangeName = "hyf.ps.exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        try (Connection cOnnection= factory.newConnection();
             Channel channel = connection.createChannel()){
            // 声明 fanout 类型的交换器
            channel.exchangeDeclare(exchangeName,"fanout");
            for (int i = 0; i <= 10; i++){
                String message = "消息"+i;
                // 直接指定交换器进行消息发布
                channel.basicPublish(exchangeName,"", null, message.getBytes());
            }
        }
    }
}

我们可以发现，我们不再需要指定 queueName，而是直接指定 exchangeName，将消息发送到交换器，由交换器决定发布到哪些 queue。

3、消费者：queue 与 exchange 建立绑定关系

建立绑定前，我们还是需要先声明 fanout 类型的交换器，并且命名要和生产者声明时的名字一致：

channel.exchangeDeclare(exchangeName, "fanout");

接着，将 queue 和 fanout 类型的交换器建立绑定消息，交换器会将消息发送到和它有绑定关系的 queue。

channel.queueBind(queueName, exchangeName, "");

此时，队列已经和交换器成功建立绑定关系，交换器接收到消息时，会发送到与交换器绑定的所有队列中。

最后，我们再调用 channel.basicConsume() 进行队列监听和绑定回调，借此来接收和消费消息：

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    System.out.println(message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });

完整代码：

public class Receive1 {

    private static final String exchangeName = "hyf.ps.exchange";
    private static final String queueName = "hyf.ps.queue1";

    public static void main(String[] args) throws IOException, TimeoutException {

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        Connection cOnnection= factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(exchangeName,"fanout");
        channel.queueDeclare(queueName,false, false, false, null);
        channel.queueBind(queueName, exchangeName,"");

        DeliverCallback callback = (s, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(message);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };

        channel.basicQos(1);
        boolean autoAck = false;
        channel.basicConsume(queueName, autoAck, callback, consumerTag -> {});
    }
}

关于发布订阅模式，我们可以理解为下图：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-e3jgCeoJ-1594103658020)(E5914A83FBA74E08865214975E25093B)]

4、发布丁订阅模式中使用工作模式

发布订阅模式中，我们还是可以继续使用上面的工作模式（多个消费者订阅同一个队列）。因为在分布式系统中，一个服务往往有多个实例，例如库存模块可以有多个实例，我们利用手动 ack 和 prefetchCount = 1，还是可以让 fanout 类型交换器的其中一个 queue 进入工作模式。

四、路由模式(routing)

上面的发布订阅模式，只要是与 fanout 类型交换器绑定的 queue，都会接收到交换器发布的消息。而我们现在的场景需要更加灵活消息分配机制。例如：error 队列只会接收到 error 类型的信息，info 队列只会接收都 info 类型的信息等等。

那么我们需要是使用灵活的路由模式，而这种模式还是需要由交换器来完成，但是此时需要使用 direct 类型的交换器来替代 fanout 类型的交换器。

bindingKey 和 routingKey

做到路由模式，不但要使用 direct 类型的交换器，还需要利用 bindingKey 和 routingKey 来完成。bindingKey 是消费者端的概念，而 routingKey 是生产者端的概念。

1、bingdingKey

发布订阅模式的消费者代码中，我们可以发现：将 queue 与交换器建立绑定关系的 queueBind() 方法中，第三个参数是空的，其实这就是配置 bindingKey 的地方。当然了，即使第三个参数不为空，fanout 类型的交换器还是会直接忽略掉的。

channel.queueBind(queueName, exchangeName, "");

例如现在我们的消费者要监听 error 类型的信息，我们需要声明 direct 类型的交换器，并且给 queue 绑定值为 error 的 bindingKey 。

public class ErrorReceive {

    private static final String exchangeName = "hyf.routing.exchange";
    private static final String queueName = "hyf.routing.error.queue";
    private static final String bindingKey = "error";

    public static void main(String[] args) throws Exception{
        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        Connection cOnnection= factory.newConnection();
        Channel channel = connection.createChannel();

        // 声明 exchange 和 queue
        channel.exchangeDeclare(exchangeName, "direct");
        channel.queueDeclare(queueName, false, false, false, null);

        // 进行绑定
        channel.queueBind(queueName, exchangeName, bindingKey);

        DeliverCallback callback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(),"utf-8");
            System.out.println("ErrorReceive 接收到" + delivery.getEnvelope().getRoutingKey() + "消息："+message);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };

        channel.basicQos(1);
        channel.basicConsume(queueName, false, callback, consumerTag -> {});
    }
}

例如现在我们的消费者2要监听 info 类型的信息，这也是非常简单，同样是上面的代码，只需要修改 queueName 和 bindingKey 即可。

// ... 省略

private static final String queueName = "hyf.info.queue";
private static final String bindingKey = "info";

// ... 省略

2、queue 绑定多个 bindingKey

上面的 hyf.error.queue 队列，只绑定了值为 error 的 bindingKey，如果现在我们不但需要接收 error 类型的信息，还需要 info 类型的信息，那么我们可以为 hyf.error.queue 再绑定多一个值为 info 的 bindingKey。

private static final String bindingKey = "error";
private static final String bindingKey2 = "info";

// 进行绑定
channel.queueBind(queueName, exchangeName, bindingKey);
channel.queueBind(queueName, exchangeName, bindingKey2);

此时，hyf.error.queue 队列同时绑定了 error 和 info 这两个 bindingKey，那么它就能同时接收到 error 类型和 info 类型的信息。

3、routingKey

在发布订阅模式中。我们可以看到发布消息的 basicPublish() 方法的第二参数是空的，而第二个参数其实就是 routingKey。

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

我们可以发现，在普通队列和工作模式中，我们都是指定 queueName 去发送消息，而 queueName 在 basicPublish 也是第二个位置。所以，在我们不使用交换器时，routingKey 指定的就是 queueName。而当我们使用交换器时，那么 routingKey 就有更丰富的含义了，它不再只是简单直接的 queueName，而是各种各样的路由含义。

要使得上面绑定了 bindingKey 为 error 和 info 的 hyf.error.queue 队列接收到消息，那么需要消息发送者指定 routingKey 为 error 或 info ，然后使用 direct 类型的交换器发布消息。

private static final String exchangeName = "hyf.log.exchange";
private static final String routingKey = "error";
private static final String routingKey2 = "info";

channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
channel.basicPublish(exchangeName, routingKey2, null, message.getBytes());

当执行上面代码，hyf.error.queue 队列能收到两条消息，而 hyf.info.queue 只能收到 routingKey 为 info 的消息。

即当 queue 绑定的 bindingKey 和发送消息时的 routingKey 完全一致，那么 queue 就能接收到交换器发送的消息，我们可以理解为下图：
在这里插入图片描述

五、主题模式(topic)

上面的路由模式虽然能让我们根据业务更加灵活的去接收指定（多种）类型的消息；但是我们可以发现，如果现在我们想让消费者接收所有类型的信息，例如 error、info、debug、fail 等消息全部都要接收，那么就要调用多次 queueBind() 方法给 queue 绑定多个 bindingKey，这就显得有点麻烦了。

此时我们可以使用主题模式，即使用 topic 类型的交换器，然后利用 * 和 # 这两个符号来搞定上面的需求。

1、* 和 # 的使用

"*" 表示匹配一个字符，"#" 表示匹配0个或多个字符

2、场景

我们现在有多个 routingKey 的消息，例如用户登陆信息 user.login.info，订单信息 order.detail.info，用户的注册信息 user.register.info，库存信息stock.detail.info 等等。

3、消费者

假设消费者1想读取到所有关于用户的信息，例如登陆信息和注册时心，那么我们可以使用 topic 类型的交换器，并且将 bindingKey 设置为 user.#。

public class UserReceive {
    private static final String exchangeName = "hyf.topic.exchange";
    private static final String bindingKey = "user.#";
    private static final String queueName = "hyf.topic.user.queue";

    @SneakyThrows
    public static void main(String[] args){

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        Connection cOnnection= factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(exchangeName, "topic");
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, exchangeName, bindingKey);

        DeliverCallback callBack = (consumerTag, delivery) -> {
            String msg = new String(delivery.getBody(), "utf-8");
            System.out.println("接收到一条user消息："+msg);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        channel.basicQos(1);
        channel.basicConsume(queueName, false, callBack, consumerTag -> {});
    }
}

假设消费者2 要接收所有上面关于信息的消息，那么他的 bindingKey 可以设置为 *.*.info。

public class InfoReceive {

    private static final String exchangeName = "hyf.topic.exchange";
    private static final String bindingKey = "*.*.info";
    private static final String queueName = "hyf.topic.info.queue";

    @SneakyThrows
    public static void main(String[] args){

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        Connection cOnnection= factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(exchangeName, "topic");
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, exchangeName, bindingKey);

        DeliverCallback callback = (consumerTag, delivery) -> {
            String msg = new String(delivery.getBody(), "utf-8");
            System.out.println("接收到一条info消息："+msg);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };

        channel.basicQos(1);
        channel.basicConsume(queueName, false, callback, consumerTag -> {});
    }
}

4、生产者

生产者也需要使用 topic 类型的交换器发送消息。

public class Send {

    private static final String exchangeName = "hyf.topic.exchange";
    private static final String routingkeyByLogin = "user.login.info";
    private static final String routingkeyByRegister = "user.register.info";
    private static final String routingkeyByOrder = "order.detail.info";
    private static final String routingkeyByStock = "stock.detail.info";

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

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        try (Connection cOnnection= factory.newConnection();
             Channel channel = connection.createChannel()){
            channel.exchangeDeclare(exchangeName, "topic");

            String msg1 = "用户张三登陆了";
            String msg2 = "新用户李四注册了";
            String msg3 = "张三买了一台iphone12";
            String msg4 = "iphone12库存减一";

            channel.basicPublish(exchangeName, routingkeyByLogin, null, msg1.getBytes());
            channel.basicPublish(exchangeName, routingkeyByRegister, null, msg2.getBytes());
            channel.basicPublish(exchangeName, routingkeyByOrder, null, msg3.getBytes());
            channel.basicPublish(exchangeName, routingkeyByStock, null, msg4.getBytes());
        }
    }
}

经过上面的代码发布消息，消费者1就能读取到消息 msg1、msg2；而消费者2可以读取到所有的消息。

关于主题模式，大家可以理解为下图：
在这里插入图片描述

六、RPC 模式

正常用 MQ 都是用来做异步化，但是有些场景却需要同步。即当我们使用 channel 发送消息后，我们需要同步等待消费者对消息消费后的结果。

RPC 模式主要是利用 replyQueue 和 correlationId 来完成。

1、客户端

客户端往 requestQueue 发送消息时需要设置 replyQueue，之后我们需要给 replyQueue 绑定一个 DeliverCallback。

为了保证客户端是同步阻塞等待结果，所以我们在 DeliverCallback 的 handle 方法里面，将结果放进阻塞队列（例如 ArrayBlockingQueue）；在代码的最后调用阻塞队列的 take() 方法在获取结果。

public class Client {

    private static final String replyQueueName = "hyf.rpc.reply.queue";
    private static final String requestQueueName = "hyf.rpc.request.queue";

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

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        Connection cOnnection= factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(replyQueueName, false, false, false, null);
        // 阻塞队列
        final BlockingQueue respOnseQueue= new ArrayBlockingQueue<>(1);

        final String corrId = UUID.randomUUID().toString();
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                .replyTo(replyQueueName)
                .correlationId(corrId)
                .build();
        String msg = "客户端消息";
        channel.basicPublish("", requestQueueName, properties, msg.getBytes());


        String ctag = channel.basicConsume(replyQueueName, true, (consumeTag,delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                responseQueue.offer(new String(delivery.getBody(), "UTF-8"));
            }
        }, consumeTag -> {});

        String result = responseQueue.take();
        System.out.println(result);
        // 取消订阅
        channel.basicCancel(ctag);
    }
}

通过上面代码，我们应该可以留意到 correlationId 的意义是什么。利用 correlationId ，我们可以判断当前从 replyQueue 获取的响应消息是否是我们发出的消息消费后的结果，如果不是我们可以直接忽略掉，保证只会获取 correlationId 一致的结果。

2、服务端

服务端在 DeilverCallback 的 handle() 方法里读取 requestQueue 里面的消息消费后，在手动 ack（关闭了自动 ack）前，需要先拿到消息的 replyQueue，然后往 replyQueue 里面发送消息消费后的结果，当然了，还要记得设置回消息的 correlatinId，最后记得手动 ack。

public class Server {

    private static final String requestQueueName = "hyf.rpc.request.queue";

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

        ConnectionFactory factory = ConnectionFactoryUtils.getFactory();
        Connection cOnnection= factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(requestQueueName, false, false, false, null);
        DeliverCallback callback = (consumerTag, delivery) -> {
            String msg = new String(delivery.getBody(), "utf-8");
            // 处理消息
            String repOnse= handleMsg(msg);
            // 将消息的 correlationId 传回去
            AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                    .Builder()
                    .correlationId(delivery.getProperties().getCorrelationId())
                    .build();
            channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, reponse.getBytes());
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };

        channel.basicQos(1);
        channel.basicConsume(requestQueueName, false, callback, consumeTag -> {});

    }

    private static String handleMsg(String msg){
        return msg + "已经被处理了";
    }
}

关于 RPC 模式，大家可以理解为下图：
在这里插入图片描述

七、总结

到此，关于 RabbitMQ 的六大使用模式已经介绍完毕。当然了，这些都是入门级别的 demo，如果大家还是有啥不明白的，可以到我的 github 上去看看，完整的代码都放在：MQ Demo。后续，我将会继续深入学习 RabbitMQ 的 Java Client，学习如何优化客户端的使用性能。