MessageQueue --- RabbitMQ
RabbitMQ IntroRabbitMQ 核心概念RabbitMQ 分发类型Dead letter (死信)保证消息的可靠传递
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: agreementsQueue A name: berlin_agreementsBinding key: agreements.eu.berlin.#Example of message routing key that matches:
agreements.eu.berlinagreements.eu.berlin.store
Consumer B is interested in all the agreements.Exchange: agreementsQueue B name: all_agreementsBinding key: agreements.#Example of message routing key that matches:
agreements.eu.berlinagreements.us
Consumer C is interested in all agreements for European head stores.Exchange: agreementsQueue C name: store_agreementsbinding key: agreements.eu.*.storeExample of message routing keys that will match:
agreements.eu.berlin.storeagreements.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 示例代码
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
public class DirectExchangeExample {
private static final String EXCHANGE_NAME = "direct_logs";
private static final String QUEUE_NAME = "my_queue";
private static final String ROUTING_KEY = "info";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
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 = new String(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）是一个特殊的队列，用于接收死信消息。一旦消息被发送到死信队列，就可以根据需要进行进一步的处理，例如重新投递、持久化、记录日志或者进行分析。
使用死信机制的好处包括：
错误处理：当消息无法被正常处理时，可以将其发送到死信队列，以便进一步处理错误情况，例如记录日志或者通知管理员。重试机制：如果消息在一定时间内未能被消费成功，可以将其发送到死信队列，并设置重试策略，例如延时重试或者指数退避重试。延迟消息：通过结合延迟队列和死信队列，可以实现延迟消息的功能。当消息的延迟时间到达时，将其发送到死信队列，然后再从死信队列中重新投递到目标队列，实现延迟消息的效果。
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
public class DeadLetterExample {
private static final String EXCHANGE_NAME = "normal_exchange";
private static final String QUEUE_NAME = "normal_queue";
private static final String DLX_EXCHANGE_NAME = "dlx_exchange";
private static final String DLX_QUEUE_NAME = "dlx_queue";
private static final String DLX_ROUTING_KEY = "dlx_routing_key";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
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
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 = new String(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
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReliableMessagingExample {
private static final String QUEUE_NAME = "my_queue";
private static final String EXCHANGE_NAME = "my_exchange";
private static final String ROUTING_KEY = "my_routing_key";
public static void main(String[] args) throws IOException, TimeoutException {
// 创建连接和信道
ConnectionFactory factory = new ConnectionFactory();
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(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.out.println("Message confirmed, delivery tag: " + deliveryTag);
}
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
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
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReliableMessagingExample {
private static final String QUEUE_NAME = "my_queue";
private static final String EXCHANGE_NAME = "my_exchange";
private static final String ROUTING_KEY = "my_routing_key";
public static void main(String[] args) throws IOException, TimeoutException {
// 创建连接和信道
ConnectionFactory factory = new ConnectionFactory();
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();
}
}
}
文章来源
您阅读本篇文章共花了：
发表评论