rabbitmq 的消息是怎么发送的？

RabbitMQ消息发送的核心组件包括生产者、连接、信道、交换机、队列和绑定。生产者通过连接建立信道，将消息发布到交换机，交换机根据类型和路由键将消息路由至队列，消费者从队列中获取消息。交换机是消息路由的“指挥官”，主要有四种类型：Direct Exchange（直连交换机）精确匹配路由键与绑定键，适用于点对点通信；Fanout Exchange（扇形交换机）广播消息到所有绑定队列，适合通知类场景；Topic Exchange（主题交换机）支持通配符模式匹配，适用于复杂路由需求；Headers Exchange（头部交换机）根据消息头部属性进行路由，灵活性高但使用较少。此外，默认交换机为直连类型，可直接将消息路由到同名队列。为确保消息可靠性，RabbitMQ提供持久化和确认机制。持久化包括交换机、队列和消息的持久化，需在声明时设置durable为true，并将消息的delivery_mode设为2，以防止服务重启导致数据丢失。确认机制分为生产者确认和消费者确认：生产者启用confirmSelect后，通过异步或同步方式接收Broker的ACK/NACK，确保消息成功送达；消费者应使用手动确认（manual-ack），在处理完成后显式发送ACK，避免因崩溃导致消息丢失。综合使用持久化与确认

RabbitMQ 消息的发送，核心来说，就是生产者（Producer）通过一个信道（Channel）将消息发布到一个交换机（Exchange），交换机再根据预设的规则将消息路由到一个或多个队列（Queue），最终等待消费者（Consumer）从这些队列中拉取或接收消息。

说白了，RabbitMQ 消息的发送过程，在我看来，其实就是生产者把消息扔到一个叫“交换机”的地方，然后交换机再根据一些规则，把它转发给合适的“队列”，最后消费者从队列里取走。这听起来可能有点抽象，但实际操作起来，核心步骤就那么几步，没那么玄乎。

生产者首先得跟 RabbitMQ 服务器建立一个连接（Connection），这就像是你要打电话，得先拨通对方的号码。连接建立后，我们会在这个连接上开辟一个或多个信道（Channel）。这个信道才是真正进行通信的“管道”，所有的消息发布、接收、队列声明、交换机声明等操作，都是通过信道来完成的。我总觉得信道就像是连接里的一条条高速公路，一条连接上可以跑很多辆车（信道），互不干扰，效率很高。

接着，生产者会声明一个交换机（Exchange），或者使用一个已经存在的交换机。交换机是消息路由的关键，它不存储消息，只负责接收消息并根据路由键（Routing Key）和自身的类型，决定把消息发往哪个队列。声明队列（Queue）也是必不可少的一步，这是消息最终的落脚点。队列和交换机之间需要通过绑定（Binding）来建立联系，告诉交换机：“嘿，如果收到符合某个条件的消息，就往我这个队列里发。”

最后，生产者通过信道将消息连同路由键一起发送给交换机。交换机收到消息后，会根据路由键和绑定规则，将消息准确无误地投递到对应的队列中。如果消息无法被路由到任何队列（比如没有匹配的绑定），那么它可能会被丢弃，或者根据交换机的配置（比如设置了 mandatory 标志）返回给生产者。

RabbitMQ 消息发送中的核心组件有哪些？

在我看来，理解 RabbitMQ 消息发送，绕不开几个核心组件，它们就像是流水线上的不同工位，各司其职，又紧密协作。

首先是生产者（Producer），这没什么好说的，就是发送消息的应用程序。它不关心消息最终去哪儿，只负责把消息“扔”出去。

然后是连接（Connection），这是生产者与 RabbitMQ Broker 之间的一条 TCP 连接。它是一个相对重量级的对象，通常在一个应用程序生命周期内只建立一次或少数几次。我个人觉得，把它想象成一条跨越海洋的光缆，建立起来耗时耗力，但一旦建成，后续的数据传输就非常便捷了。

接着是信道（Channel），这是在连接内部建立的轻量级逻辑连接。所有的消息发布和接收都是通过信道进行的。一个连接可以有多个信道，而且信道之间是相互独立的，可以并行操作。这设计真的很有意思，它避免了为每个操作都建立一个重量级的 TCP 连接，大大提升了性能和资源利用率。我经常在想，如果每次操作都要建连接，那系统资源得多浪费啊。

再来是交换机（Exchange），这是消息路由的第一站。生产者把消息发给它，它根据类型和路由键把消息分发到不同的队列。交换机本身不存储消息，只负责转发。它就像一个交通枢纽，根据车辆的目的地（路由键）和自身的规则（类型），指引车辆走向正确的道路。

然后是队列（Queue），这是消息的最终目的地，也是消费者获取消息的地方。队列是有序的，先进先出。它就像一个邮局的信箱，消息在这里排队等待被消费者取走。

最后是绑定（Binding），这是连接交换机和队列的桥梁。它定义了交换机如何根据路由键将消息路由到队列。一个队列可以绑定到多个交换机，一个交换机也可以绑定到多个队列。

Exchange 在 RabbitMQ 消息路由中扮演什么角色？它有哪几种类型？

交换机在 RabbitMQ 消息路由中扮演着至关重要的“指挥官”角色。它不存储消息，而是根据自身类型和消息的路由键（Routing Key），决定将消息投递到哪些队列。简单来说，它就是消息的第一个“分拣中心”。

RabbitMQ 主要提供了四种类型的交换机，每种都有其特定的路由行为：

1. Direct Exchange（直连交换机）：

   

   智谱清言 - 免费全能的AI助手

   智谱清言 - 免费全能的AI助手

   2

   查看详情
   • 路由规则：它会精确匹配消息的路由键和队列的绑定键（Binding Key）。只有当消息的路由键与队列的绑定键完全一致时，消息才会被投递到该队列。
   • 使用场景：适用于点对点消息传递，或者需要精确控制消息流向的场景。比如，你想把特定类型的日志消息发送给特定的处理服务。
   • 举个例子：一个交换机 log_exchange，一个队列 error_queue 绑定了 error 路由键，另一个队列 info_queue 绑定了 info 路由键。如果生产者发送一个路由键为 error 的消息，它就只会进入 error_queue。

2. Fanout Exchange（扇形交换机）：

   • 路由规则：它会把收到的所有消息广播给所有绑定到它的队列，忽略消息的路由键。
   • 使用场景：适用于广播消息、通知所有订阅者或负载均衡等场景。
   • 举个例子：一个交换机 notification_exchange，无论发送什么路由键的消息，只要有队列绑定了它，这个队列就能收到消息。就像你发了一条朋友圈，所有朋友都能看到。

3. Topic Exchange（主题交换机）：

   • 路由规则：这是最灵活的一种。它通过模式匹配的方式来路由消息，路由键和绑定键都是由点号分隔的单词列表。绑定键中可以使用通配符：*（匹配一个单词）和 #（匹配零个或多个单词）。
   • 使用场景：适用于日志系统、新闻订阅、多播等复杂的路由场景。
   • 举个例子：一个交换机 news_exchange，一个队列 us.sports.queue 绑定了 us.sports.*，另一个队列 all.news.queue 绑定了 #.news。
     • 如果消息路由键是 us.sports.basketball，它会进入 us.sports.queue 和 all.news.queue。
     • 如果消息路由键是 eu.politics.news，它只会进入 all.news.queue。

4. Headers Exchange（头部交换机）：

   • 路由规则：它不依赖路由键，而是根据消息的头部属性（Headers）进行匹配。绑定时需要指定一个 x-match 参数（可以是 all 或 any），表示所有头部都匹配才路由，还是任意一个匹配就路由。
   • 使用场景：当路由逻辑与消息内容或元数据紧密相关，且路由键无法满足需求时。相对而言，用得比较少，因为它比 Topic 更复杂些。

此外，还有个默认交换机（Default Exchange），它是一个直连交换机，没有名字（空字符串 ""）。当你发送消息时没有指定交换机，消息就会被发送到这个默认交换机。它会根据消息的路由键，直接投递到同名的队列中。这在简单场景下非常方便，省去了声明交换机和绑定的步骤。

消息发送时如何确保可靠性？持久化和确认机制怎么用？

在分布式系统中，消息的可靠性是重中之重，谁也不想消息发出去就石沉大海，或者因为系统崩溃而丢失。RabbitMQ 提供了多种机制来确保消息在发送过程中的可靠性，主要就是持久化和确认机制。

持久化（Durability）：

持久化是为了防止 RabbitMQ 服务重启或崩溃时消息丢失。它涉及三个层面的持久化：

1. 交换机持久化（Durable Exchange）：

   • 声明交换机时，可以将其设置为 durable。这样即使 RabbitMQ 服务重启，这个交换机也会被重新创建。
   • 代码里通常是 channel.exchangeDeclare("my_exchange", "direct", true); 这里的 true 就是持久化。
   • 我个人觉得，除非是临时性交换机，否则大部分生产环境的交换机都应该设置为持久化，以防万一。

2. 队列持久化（Durable Queue）：

   • 声明队列时，也可以将其设置为 durable。这意味着队列的元数据（队列名、属性等）以及其中存储的消息（如果消息本身也是持久化的）会在 RabbitMQ 重启后依然存在。
   • 代码通常是 channel.queueDeclare("my_queue", true, false, false, null); 这里的第二个 true 就是持久化。
   • 注意，如果一个持久化的队列绑定了一个非持久化的交换机，或者反之，这并不会影响各自的持久性。但为了整体可靠性，通常会保持一致。

3. 消息持久化（Persistent Messages）：

   • 这是最关键的一环。即使交换机和队列都持久化了，如果消息本身不是持久化的，那么在 RabbitMQ 重启时，队列中尚未被消费的消息依然会丢失。
   • 发送消息时，需要将消息的 delivery_mode 设置为 2（表示持久化）。
   • 代码示例：channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
   • 消息持久化会将消息写入磁盘。这会带来一些 I/O 开销，降低吞吐量，但换来的是数据安全。所以，你需要根据业务场景权衡。对于那些“绝对不能丢”的消息，持久化是必须的。

确认机制（Acknowledgments）：

确认机制确保消息在发送和消费过程中不丢失，它分为生产者确认和消费者确认。

1. 生产者确认（Publisher Confirms）：

   • 这解决了消息从生产者发送到 RabbitMQ Broker 过程中可能丢失的问题。生产者发送消息后，Broker 会给生产者一个确认（ACK），表示消息已成功接收并处理（通常是已写入磁盘或路由到队列）。
   • 有两种模式：
     • 同步确认：生产者发送一条消息后，等待 Broker 的确认，收到确认后再发送下一条。这种方式简单，但吞吐量低，因为它阻塞了发送。
     • 异步确认：生产者发送消息后不等待，继续发送下一条。Broker 会异步地通过回调通知生产者哪些消息已确认，哪些未确认（NACK）。未确认的消息需要生产者进行重试。这是生产环境推荐的方式，兼顾了可靠性和性能。
   • 启用方式：channel.confirmSelect(); 然后监听 ConfirmListener 或使用 waitForConfirms()。
   • 我个人更倾向于异步确认，虽然代码复杂一点，但性能提升是实实在在的。同步确认只适合对吞吐量要求不高的场景，或者作为快速验证的手段。

2. 消费者确认（Consumer Acknowledgments）：

   • 这解决了消息从队列投递给消费者后，在消费者处理过程中可能丢失或未处理的问题。
   • 当消费者收到消息并成功处理后，需要向 RabbitMQ 发送一个确认（ACK）。RabbitMQ 收到 ACK 后，才会将该消息从队列中删除。
   • 如果消费者在处理消息过程中崩溃，或者没有发送 ACK，那么 RabbitMQ 会认为消息未被成功处理，会将消息重新投递给其他消费者（如果存在），或者在消费者恢复后重新投递给它。
   • 确认模式：
     • 自动确认（Auto-ACK）：消息一旦被投递给消费者，就立即从队列中删除。这最简单，但风险最高，如果消费者处理失败，消息就丢了。
     • 手动确认（Manual-ACK）：消费者处理完消息后，显式地发送 ACK。这是生产环境推荐的方式，提供了最大的灵活性和可靠性。
       • channel.basicAck(deliveryTag, multiple)：确认消息。deliveryTag 是消息的唯一标识，multiple 为 true 表示确认 deliveryTag 之前所有未确认的消息。
       • channel.basicNack(deliveryTag, multiple, requeue)：拒绝消息，可以选择是否重新入队。
       • channel.basicReject(deliveryTag, requeue)：拒绝单条消息，功能与 basicNack 类似，但不能批量。
   • 在我的经验里，手动确认几乎是所有关键业务的标配。虽然处理起来稍微麻烦点，但能有效避免消息丢失，这是最基础的保障。特别是在处理过程中可能出现异常的业务逻辑里，手动确认配合重试机制，简直是消息可靠性的“双保险”。

以上就是rabbitmq 的消息是怎么发送的？的详细内容，更多请关注php中文网其它相关文章！