MessageQueue --- RabbitMQ

MessageQueue --- RabbitMQ

RabbitMQ Intro

• 2007年发布，是一个在AMQP（高级消息队列协议）基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。
• 低延迟：RabbitMQ 提供了低延迟的消息传递，可以在毫秒级别内将消息从生产者传递到消费者。
• 高吞吐量：RabbitMQ 能够处理大量的消息并实现高吞吐量。它使用多线程和预取机制来提高消息处理的效率。
• 可扩展性：RabbitMQ 可以通过水平扩展来处理更多的消息流量。可以通过添加更多的节点、使用集群和队列分区等方式来扩展 RabbitMQ。
• 持久化支持：RabbitMQ 支持将消息和队列持久化到磁盘，以确保消息的可靠性和持久性。这意味着即使在 RabbitMQ 重启后，消息仍然可以保留，不会丢失。
• 多种消息传递模式：RabbitMQ 支持多种消息传递模式，如点对点、发布/订阅和请求/响应等。这使得 RabbitMQ 在各种场景下都能够灵活应用。
• 负载均衡：RabbitMQ 提供了负载均衡机制，可以将消息均衡地分发给多个消费者，以实现更好的资源利用和处理能力。
• 可靠性保证：通过使用确认机制、持久化和事务等特性，RabbitMQ 提供了可靠性保证，确保消息的可靠传递和处理。

RabbitMQ 核心概念

消息的路由过程如下：

• 生产者发布消息时，将消息和指定的 Routing Key 一起发送到交换机。
• 交换机根据 Binding Key 和 Routing Key 的匹配规则，将消息路由到一个或多个绑定的队列。
• 绑定 Key 和 Routing Key 的匹配规则可以根据交换机的类型而有所不同。

• 在直接交换机（Direct Exchange）中，Binding Key 必须与 Routing Key 完全匹配。
• 在主题交换机（Topic Exchange）中，Binding Key 与 Routing Key 使用通配符进行模式匹配。
• 在扇形交换机（Fanout Exchange）中，Binding Key 不起作用，消息会被广播到所有绑定的队列。

• Routing Key（路由键）：

• Routing Key 是在消息发布时与消息一起发送的属性。
• 在发布消息时，生产者可以指定一个 Routing Key，用于描述消息的特性或目标。
• Routing Key 可以是任意的字符串，通常是由一些特定的标识符或关键词组成，以便用于消息的过滤和路由。

• Binding Key（绑定键）：

• Binding Key 是用于绑定队列和交换机的属性。
• 在 RabbitMQ 中，通过创建绑定（Binding）将队列和交换机关联起来，以便消息能够正确地路由到队列。
• 绑定是基于 Binding Key 进行的，它定义了交换机将消息路由到哪些队列。
• Binding Key 可以是一个或多个单词组成的字符串，也可以是符号“#”和“*”的组合，用于匹配 Routing Key。

名词解释：

• 生产者（Producer）：发送消息的应用。
• 消费者（Consumer）：接收消息的应用。
• 队列（Queue）：存储消息的缓存。
• 消息（Message）：由生产者通过RabbitMQ发送给消费者的信息。
• 连接（Connection）：连接RabbitMQ和应用服务器的TCP连接。
• 信道（Channel）：连接里的一个虚拟通道，通过消息队列发送或者接收消息时，都是通过信道进行的。
• 交换机（Exchange）：交换机负责从生产者那里接收消息，并根据交换类型分发到对应的消息队列里。
• 代理（Broker）：接收和分发消息的应用，RabbitMQ Server就是Message Broker。
• 虚拟主机（Virtual host）：出于多租户和安全因素设计的，把AMQP的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建exchange/queue 等.

• Example：每个环境配置一个virtual host

RabbitMQ 分发类型

Direct Exchange

• Direct exchange使用完全匹配的方式进行消息路由

• 当一个消息发送到Direct exchange时，它会将消息的路由键与绑定到交换机上的队列的绑定键（binding key）进行比较。如果路由键与某个队列的绑定键完全匹配，那么该消息将被路由到该队列

• exchange ：pdf_events Queue A：create_pdf_queue 交换机（pdf_events）和队列 A（create_pdf_queue）之间的绑定键：pdf_create

• exchange ：pdf_events Queue B：pdf_log_queue 交换机（pdf_events）和队列 B（pdf_log_queue）之间的绑定键：pdf_log

• 示例：

• 例如，一个带有routing key为 pdf_log 的消息被发送到交换机 pdf_events
• 该消息会被路由到 pdf_log_queue，因为routing key（pdf_log）与binding key（pdf_log）匹配
• 如果消息的路由键与任何绑定键都不匹配，那么该消息将被丢弃。

Topic Exchange

• 在 Topic Exchange 中，消息的路由键和绑定键都使用通配符来进行匹配。路由键是消息的属性，而绑定键是在创建绑定时指定的。通配符可以帮助实现更灵活的消息路由，使得消息可以根据特定的模式进行匹配和分发。
• 通配符符号：

• *（星号）：表示匹配一个单词（单词由点号分隔）
• #（井号）：表示匹配零个或多个单词（单词由点号分隔）

• Consumer A is interested in all the agreements in Berlin.
• Exchange: agreements
• Queue A name: berlin_agreements
• Binding key: agreements.eu.berlin.#
• Example of message routing key that matches:

• agreements.eu.berlin
• agreements.eu.berlin.store

• Consumer B is interested in all the agreements.
• Exchange: agreements
• Queue B name: all_agreements
• Binding key: agreements.#
• Example of message routing key that matches:

• agreements.eu.berlin
• agreements.us

• Consumer C is interested in all agreements for European head stores.
• Exchange: agreements
• Queue C name: store_agreements
• binding key: agreements.eu.*.store
• Example of message routing keys that will match:

• agreements.eu.berlin.store
• agreements.eu.stockholm.store

Fanout exchange

• Fanout Exchange（广播交换机）是 RabbitMQ 中的一种交换机类型。它的工作原理是将消息广播到与之绑定的所有队列，无论绑定键的匹配情况如何。
• Fanout Exchange 不关心消息的路由键，它会简单地将收到的消息复制并发送到所有与之绑定的队列

Header exchange

• Header Exchange（头交换机）是 RabbitMQ 中的一种交换机类型。它使用消息的头部属性（Header）来匹配与之绑定的队列，而不依赖于路由键或绑定键。

• 在 Header Exchange 中，消息的头部属性是消息的一部分，它包含了一组键值对，用于描述消息的特征和属性。与其他类型的交换机不同，Header Exchange 不关心消息的路由键，而是根据消息头部属性的匹配情况来确定消息的路由。

• 绑定到 Header Exchange 的队列可以定义一个或多个匹配规则，这些规则由键值对的匹配条件组成。当消息的头部属性与队列的匹配规则完全匹配时，消息会被路由到对应的队列。

• Message 1 is published to the exchange with header arguments (key = value): “format = pdf”, “type = report”.
• Message 1 is delivered to Queue A because all key/value pairs match, and Queue B since “format = pdf” is a match (binding rule set to “x-match =any”).

• Message 2 is published to the exchange with header arguments of (key = value): “format = pdf”.
• Message 2 is only delivered to Queue B. Because the binding of Queue A requires both “format = pdf” and “type = report” while Queue B is configured to match any key-value pair (x-match = any) as long as either “format = pdf” or “type = log” is present.

• Message 3 is published to the exchange with header arguments of (key = value): “format = zip”, “type = log”.
• Message 3 is delivered to Queue B since its binding indicates that it accepts messages with the key-value pair “type = log”, it doesn’t mind that “format = zip” since “x-match = any”.
• Queue C doesn’t receive any of the messages since its binding is configured to match all of the headers (“x-match = all”) with “format = zip”, “type = pdf”. No message in this example lives up to these criterias.
• It’s worth noting that in a header exchange, the actual order of the key-value pairs in the message is irrelevant.

配置RabbitMQ 示例代码

importcom.rabbitmq.client.*;importjava.io.IOException;importjava.nio.charset.StandardCharsets;publicclassDirectExchangeExample{privatestaticfinalStringEXCHANGE_NAME="direct_logs";privatestaticfinalStringQUEUE_NAME="my_queue";privatestaticfinalStringROUTING_KEY="info";publicstaticvoidmain(String[] args)throwsException{ConnectionFactory factory =newConnectionFactory();
        factory.setHost("localhost");try(Connection connection = factory.newConnection();Channel channel = connection.createChannel()){// 声明一个 Direct Exchange
            channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT);// 声明一个队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);// 绑定队列到 Direct Exchange，并指定绑定键
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);// 定义消息处理函数DeliverCallback deliverCallback =(consumerTag, delivery)->{String message =newString(delivery.getBody(),StandardCharsets.UTF_8);System.out.println("Received message: "+ message);};// 消费消息
            channel.basicConsume(QUEUE_NAME,true, deliverCallback, consumerTag ->{});System.out.println("Press any key to exit.");System.in.read();}}}

Dead letter (死信)

• 死信（Dead Letter）是指在消息队列中无法被正常消费和处理的消息。当消息满足一定的条件时,它们可以被标记为死信并被发送到专门的死信队列中，以便进一步处理或分析
• 死信来源

• 消息 TTL 过期
• 队列达到最大长度（队列满了，无法再添加数据到 mq 中）
• 消息被拒绝（basic.reject 或 basic.nack）并且 requeue=false（不再重新入队）

• 死信队列（Dead Letter Queue）是一个特殊的队列，用于接收死信消息。一旦消息被发送到死信队列，就可以根据需要进行进一步的处理，例如重新投递、持久化、记录日志或者进行分析。

• 使用死信机制的好处包括：

• 错误处理：当消息无法被正常处理时，可以将其发送到死信队列，以便进一步处理错误情况，例如记录日志或者通知管理员。
• 重试机制：如果消息在一定时间内未能被消费成功，可以将其发送到死信队列，并设置重试策略，例如延时重试或者指数退避重试。
• 延迟消息：通过结合延迟队列和死信队列，可以实现延迟消息的功能。当消息的延迟时间到达时，将其发送到死信队列，然后再从死信队列中重新投递到目标队列，实现延迟消息的效果。

importcom.rabbitmq.client.*;importjava.io.IOException;importjava.nio.charset.StandardCharsets;importjava.util.HashMap;importjava.util.Map;publicclassDeadLetterExample{privatestaticfinalStringEXCHANGE_NAME="normal_exchange";privatestaticfinalStringQUEUE_NAME="normal_queue";privatestaticfinalStringDLX_EXCHANGE_NAME="dlx_exchange";privatestaticfinalStringDLX_QUEUE_NAME="dlx_queue";privatestaticfinalStringDLX_ROUTING_KEY="dlx_routing_key";publicstaticvoidmain(String[] args)throwsException{ConnectionFactory factory =newConnectionFactory();
        factory.setHost("localhost");try(Connection connection = factory.newConnection();Channel channel = connection.createChannel()){// 创建普通交换机和队列
            channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");// 创建死信交换机和队列
            channel.exchangeDeclare(DLX_EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
            channel.queueDeclare(DLX_QUEUE_NAME,false,false,false,null);
            channel.queueBind(DLX_QUEUE_NAME,DLX_EXCHANGE_NAME,DLX_ROUTING_KEY);// 设置普通队列的死信参数Map<String,Object> arguments =newHashMap<>();
            arguments.put("x-dead-letter-exchange",DLX_EXCHANGE_NAME);
            arguments.put("x-dead-letter-routing-key",DLX_ROUTING_KEY);
            channel.queueDeclare(QUEUE_NAME,false,false,false, arguments);// 定义消息处理函数DeliverCallback deliverCallback =(consumerTag, delivery)->{String message =newString(delivery.getBody(),StandardCharsets.UTF_8);System.out.println("Received message: "+ message);// 手动确认消息
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);};// 消费消息
            channel.basicConsume(QUEUE_NAME,false, deliverCallback, consumerTag ->{});System.out.println("Press any key to exit.");System.in.read();}}}

保证消息的可靠传递

要确保消息的可靠传递，可以采取以下几个步骤：

• 持久化消息：将消息和队列都设置为持久化。这样，在 RabbitMQ 重启后，持久化的队列和消息会被恢复，避免消息丢失。可以在消息的发布端设置消息的持久化属性，以及在队列声明时设置队列的持久化属性。
• 使用确认机制（Acknowledgement）：在消费者处理消息后，发送确认消息给 RabbitMQ，告知消息已经成功处理。RabbitMQ 收到确认后才会将消息从队列中删除，确保消息不会丢失。确认机制可以通过在消费者端手动发送确认消息（basicAck）或使用自动确认模式（autoAck）来实现。
• 使用发布者确认（Publisher Confirms）：在消息的发布端启用发布者确认模式。通过将 confirm.select 设置为 true，可以让发布者等待 RabbitMQ 发送确认消息，表示消息已经成功到达交换机。如果没有收到确认消息，发布者可以选择重新发送消息，确保消息的可靠传递。
• 设置合适的消息 TTL（Time-to-Live）：可以为消息设置 TTL，即消息的存活时间。如果消息在指定的时间内没有被消费，RabbitMQ 可以将其标记为过期并进行相应的处理，例如发送到死信队列或丢弃。
• 使用事务（Transactions）：可以通过开启事务来确保消息的可靠传递。在事务中，可以将消息的发布和确认操作包裹在一个事务中，如果事务提交成功，表示消息已经成功到达 RabbitMQ，否则可以进行回滚。
• 备份交换机（Alternate Exchange）：可以配置备份交换机，当消息无法路由到指定的交换机时，它将被发送到备份交换机，从而避免消息丢失。
• 监控和错误处理：建立监控机制，定期检查消息队列的状态，以及消费者的消费情况。在出现错误或异常情况时，根据具体情况进行错误处理，例如重试发送消息、记录日志、发送警报等。

发布者确认机制Example

importcom.rabbitmq.client.*;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassReliableMessagingExample{privatestaticfinalStringQUEUE_NAME="my_queue";privatestaticfinalStringEXCHANGE_NAME="my_exchange";privatestaticfinalStringROUTING_KEY="my_routing_key";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{// 创建连接和信道ConnectionFactory factory =newConnectionFactory();
        factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明队列和交换机
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);// 启用发布者确认模式
        channel.confirmSelect();// 添加发布者确认监听器
        channel.addConfirmListener(newConfirmListener(){@OverridepublicvoidhandleAck(long deliveryTag,boolean multiple)throwsIOException{System.out.println("Message confirmed, delivery tag: "+ deliveryTag);}@OverridepublicvoidhandleNack(long deliveryTag,boolean multiple)throwsIOException{System.out.println("Message not confirmed, delivery tag: "+ deliveryTag);// 可以在这里进行相应的处理，例如重新发送消息}});// 发布消息String message ="Hello, RabbitMQ!";
        channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());try{// 等待发布者确认
            channel.waitForConfirmsOrDie();}catch(InterruptedException e){// 可以在这里进行相应的处理，例如重新发送消息
            e.printStackTrace();}// 关闭信道和连接
        channel.close();
        connection.close();}}

事务保证消息可靠性Example

importcom.rabbitmq.client.*;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassReliableMessagingExample{privatestaticfinalStringQUEUE_NAME="my_queue";privatestaticfinalStringEXCHANGE_NAME="my_exchange";privatestaticfinalStringROUTING_KEY="my_routing_key";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{// 创建连接和信道ConnectionFactory factory =newConnectionFactory();
        factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();try{// 开启事务
            channel.txSelect();// 声明队列和交换机
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);// 发布消息String message ="Hello, RabbitMQ!";
            channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());// 提交事务
            channel.txCommit();System.out.println("Transaction committed successfully.");}catch(IOException e){// 发生异常，回滚事务
            channel.txRollback();System.out.println("Transaction rolled back due to an exception.");
            e.printStackTrace();}finally{// 关闭信道和连接
            channel.close();
            connection.close();}}}