做自媒体查找素材的网站,高端建材门店年销售额,计算机应用技术培训班,搜索引擎优化培训免费咨询MessageQueue --- RabbitMQ RabbitMQ IntroRabbitMQ 核心概念RabbitMQ 分发类型Dead letter (死信)保证消息的可靠传递 RabbitMQ Intro 2007年发布#xff0c;是一个在AMQP#xff08;高级消息队列协议#xff09;基础上完成的#xff0c;可复用的企业消息系统#xff0c;… 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 Acreate_pdf_queue 交换机pdf_events和队列 Acreate_pdf_queue之间的绑定键pdf_create exchange pdf_events Queue Bpdf_log_queue 交换机pdf_events和队列 Bpdf_log_queue之间的绑定键pdf_log 示例 例如一个带有routing key为 pdf_log 的消息被发送到交换机 pdf_events该消息会被路由到 pdf_log_queue因为routing keypdf_log与binding keypdf_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 Exchangechannel.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并且 requeuefalse不再重新入队 死信队列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);// 设置普通队列的死信参数MapString, Object arguments new HashMap();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 发送确认消息表示消息已经成功到达交换机。如果没有收到确认消息发布者可以选择重新发送消息确保消息的可靠传递。设置合适的消息 TTLTime-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() {Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {System.out.println(Message confirmed, delivery tag: deliveryTag);}Overridepublic 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();}}
}
