RabbitMQ 消息顺序、消息幂等、消息重复、消息事务、集群

本文主要是介绍RabbitMQ 消息顺序、消息幂等、消息重复、消息事务、集群，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

1. 消息顺序

场景：比如下单操作，下单成功之后，会发布创建订单和扣减库存消息，但扣减库存消息执行会先于创建订单消息，也就说前者执行成功之后，才能执行后者。

不保证完全按照顺序消费，在 MQ 层面支持消息的顺序处理开销太大，为了极少量的需求，增加整体上的复杂度得不偿失。

所以，还是在应用层面处理比较好，或者业务逻辑进行处理。

应用层解决方式：

• 1. 消息实体中增加：版本号 & 状态机 & msgid & parent_msgid，通过 parent_msgid 判断消息的顺序（需要全局存储，记录消息的执行状态）。 2. “同步执行”：当一个消息执行完之后，再发布下一个消息。

2. 消息幂等、消息重复、消息事务

消息重复

造成消息重复的根本原因是：网络不可达。只要通过网络交换数据，就无法避免这个问题。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？

消费端处理消息的业务逻辑保持幂等性。

保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现。

第 1 条很好理解，只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样。第 2 条原理就是利用一张日志表来记录已经处理成功的消息的 ID，如果新到的消息 ID 已经在日志表中，那么就不再处理这条消息。

第 1 条解决方案，很明显应该在消费端实现，不属于消息系统要实现的功能。第 2 条可以消息系统实现，也可以业务端实现。正常情况下出现重复消息的概率其实很小，如果由消息系统来实现的话，肯定会对消息系统的吞吐量和高可用有影响，所以最好还是由业务端自己处理消息重复的问题，这也是 RabbitMQ 不解决消息重复的问题的原因。

RabbitMQ 不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重。

AMQP 消费者确认机制

AMQP 定义了消费者确认机制（message ack），如果一个消费者应用崩溃掉（此时连接会断掉，broker 会得知），但是 broker 尚未获得 ack，那么消息会被重新放入队列。所以 AMQP 提供的是“至少一次交付”（at-least-once delivery），异常情况下，消息会被重复消费，此时业务要实现幂等性（重复消息处理）。

AMQP 生产者事务

对于生产者，AMQP 定义了事务（tx transaction）来确保生产消息被 broker 接收并成功入队。TX 事务是阻塞调用，生产者需等待broker写磁盘后返回的确认，之后才能继续发送消息。事务提交失败时（如broker宕机场景），broker并不保证提交的消息全部入队。

TX 的阻塞调用使 broker 的性能非常差，RabbitMQ 使用 confirm 机制来优化生产消息的确认。Confirm 模式下，生产者可以持续发送消息，broker 将消息批量写磁盘后回复确认，生产者通过确认消息的ID来确定哪些已发送消息被成功接收。Confirm 模式下生产者发送消息和接受确认是异步流程，生产者需要缓存未确认的消息以便出错时重新发送。

总结

• 1. 消息重复发布：不存在，因为 AMQP 定义了事务（tx transaction）来确保生产消息被 broker 接收并成功入队。TX 事务是阻塞调用，生产者需等待 broker 写磁盘后返回的确认，之后才能继续发送消息。事务提交失败时（如 broker 宕机场景），broker 并不保证提交的消息全部入队。RabbitMQ 使用 confirm 机制来优化生产消息的确认（可以持续发布消息，但会批量回复确认）。 2. 消息重复消费：AMQP 提供的是“至少一次交付”（at-least-once delivery），异常情况下，消息会被重复消费，此时业务要实现幂等性（重复消息处理）。

应用层解决方式：

• 1. 专门的 Map 存储：用来存储每个消息的执行状态（用 msgid 区分），执行成功之后更新 Map，有另外消息重复消费的时候，读取 Map 数据判断 msgid 对应的执行状态，已消费则不执行。 2. 业务逻辑判断：消息执行完会更改某个实体状态，判断实体状态是否更新，如果更新，则不进行重复消费。

特别说明：AMQP 协议中的事务仅仅是指生产者发送消息给 broker 这一系列流程处理的事务机制，并不包含消费端的处理流程。

3. 集群

原 RabbitMQ 集群：manager1、manager2、manager3 节点均为磁盘存储，manager1 为主节点，HAProxy 负载三个节点。

现 RabbitMQ 集群更新（更合理的配置）：

• 1. RabbitMQ 集群更新：manager1、manager2 节点类型改为 ram（内存存储），manager3 节点类型为 disc（磁盘存储，用于保存集群配置和元数据），主节点变更为 manager3。 2. HAProxy 负载更新：移除 manager3 负载（5672 端口），只保留 manage2、manager2 负载。

4. Kafka 和 RabbitMQ 对比

Kafka 的设计有明确的介绍：。

Kafka 应对场景：消息持久化、吞吐量是第一要求、状态由客户端维护、必须是分布式的。Kafka 认为 broker 不应该阻塞生产者，高效的磁盘顺序读写能够和网络 IO 一样快，同时依赖现代 OS 文件系统特性，写入持久化文件时并不调用 flush，仅写入 OS pagecache，后续由 OS flush。

这些特性决定了 Kafka 没有做“确认机制”，而是直接将生产消息顺序写入文件、消息消费后不删除（避免文件更新），该实现充分利用了磁盘 IO，能够达到较高的吞吐量。代价是消费者要依赖 Zookeeper 记录队列消费位置、处理同步问题。没有消费确认机制，还导致了 Kafka 无法了解消费者速度，不能采用 push 模型以合理的速度向消费者推送数据，只能利用 pull 模型由消费者来拉消息（消费者承担额外的轮询开销）。

如果在 Kafka 中引入消费者确认机制，就需要 broker 维护消息消费状态，要做到高可靠就需要写文件持久化并与生产消息同步，这将急剧降低 Kafka 的性能，这种设计也极类似 RabbitMQ。如果不改变 Kafka 的实现，而是在 Kafka 和消费者之间做一层封装，还是需要实现一套类似 RabbitMQ 的消费确认和持久化机制。

参考资料：

1. 消息顺序 场景：比如下单操作，下单成功之后，会发布创建订单和扣减库存消息，但扣减库存消息执行会先于创建订单消息，也就说前者执行成功之后，才能执行后者。 不保证完全按照顺序消费，在 MQ 层面支持消息的顺序处理开销太大，为了极少量的需求，增加整体上的复杂度得不偿失。 所以，还是在应用层面处理比较好，或者业务逻辑进行处理。 应用层解决方式： 1. 消息实体中增加：版本号 & 状态机 & msgid & parent_msgid，通过 parent_msgid 判断消息的顺序（需要全局存储，记录消息的执行状态）。 2. “同步执行”：当一个消息执行完之后，再发布下一个消息。 2. 消息幂等、消息重复、消息事务 消息重复 造成消息重复的根本原因是：网络不可达。只要通过网络交换数据，就无法避免这个问题。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？ 消费端处理消息的业务逻辑保持幂等性。 保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现。 第 1 条很好理解，只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样。第 2 条原理就是利用一张日志表来记录已经处理成功的消息的 ID，如果新到的消息 ID 已经在日志表中，那么就不再处理这条消息。 第 1 条解决方案，很明显应该在消费端实现，不属于消息系统要实现的功能。第 2 条可以消息系统实现，也可以业务端实现。正常情况下出现重复消息的概率其实很小，如果由消息系统来实现的话，肯定会对消息系统的吞吐量和高可用有影响，所以最好还是由业务端自己处理消息重复的问题，这也是 RabbitMQ 不解决消息重复的问题的原因。 RabbitMQ 不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重。 AMQP 消费者确认机制 AMQP 定义了消费者确认机制（message ack），如果一个消费者应用崩溃掉（此时连接会断掉，broker 会得知），但是 broker 尚未获得 ack，那么消息会被重新放入队列。所以 AMQP 提供的是“至少一次交付”（at-least-once delivery），异常情况下，消息会被重复消费，此时业务要实现幂等性（重复消息处理）。 AMQP 生产者事务 对于生产者，AMQP 定义了事务（tx transaction）来确保生产消息被 broker 接收并成功入队。TX 事务是阻塞调用，生产者需等待broker写磁盘后返回的确认，之后才能继续发送消息。事务提交失败时（如broker宕机场景），broker并不保证提交的消息全部入队。 TX 的阻塞调用使 broker 的性能非常差，RabbitMQ 使用 confirm 机制来优化生产消息的确认。Confirm 模式下，生产者可以持续发送消息，broker 将消息批量写磁盘后回复确认，生产者通过确认消息的ID来确定哪些已发送消息被成功接收。Confirm 模式下生产者发送消息和接受确认是异步流程，生产者需要缓存未确认的消息以便出错时重新发送。 总结 1. 消息重复发布：不存在，因为 AMQP 定义了事务（tx transaction）来确保生产消息被 broker 接收并成功入队。TX 事务是阻塞调用，生产者需等待 broker 写磁盘后返回的确认，之后才能继续发送消息。事务提交失败时（如 broker 宕机场景），broker 并不保证提交的消息全部入队。RabbitMQ 使用 confirm 机制来优化生产消息的确认（可以持续发布消息，但会批量回复确认）。 2. 消息重复消费：AMQP 提供的是“至少一次交付”（at-least-once delivery），异常情况下，消息会被重复消费，此时业务要实现幂等性（重复消息处理）。 应用层解决方式： 1. 专门的 Map 存储：用来存储每个消息的执行状态（用 msgid 区分），执行成功之后更新 Map，有另外消息重复消费的时候，读取 Map 数据判断 msgid 对应的执行状态，已消费则不执行。 2. 业务逻辑判断：消息执行完会更改某个实体状态，判断实体状态是否更新，如果更新，则不进行重复消费。 特别说明：AMQP 协议中的事务仅仅是指生产者发送消息给 broker 这一系列流程处理的事务机制，并不包含消费端的处理流程。 3. 集群 原 RabbitMQ 集群：manager1、manager2、manager3 节点均为磁盘存储，manager1 为主节点，HAProxy 负载三个节点。 现 RabbitMQ 集群更新（更合理的配置）： 1. RabbitMQ 集群更新：manager1、manager2 节点类型改为 ram（内存存储），manager3 节点类型为 disc（磁盘存储，用于保存集群配置和元数据），主节点变更为 manager3。 2. HAProxy 负载更新：移除 manager3 负载（5672 端口），只保留 manage2、manager2 负载。 4. Kafka 和 RabbitMQ 对比 Kafka 的设计有明确的介绍：。 Kafka 应对场景：消息持久化、吞吐量是第一要求、状态由客户端维护、必须是分布式的。Kafka 认为 broker 不应该阻塞生产者，高效的磁盘顺序读写能够和网络 IO 一样快，同时依赖现代 OS 文件系统特性，写入持久化文件时并不调用 flush，仅写入 OS pagecache，后续由 OS flush。 这些特性决定了 Kafka 没有做“确认机制”，而是直接将生产消息顺序写入文件、消息消费后不删除（避免文件更新），该实现充分利用了磁盘 IO，能够达到较高的吞吐量。代价是消费者要依赖 Zookeeper 记录队列消费位置、处理同步问题。没有消费确认机制，还导致了 Kafka 无法了解消费者速度，不能采用 push 模型以合理的速度向消费者推送数据，只能利用 pull 模型由消费者来拉消息（消费者承担额外的轮询开销）。 如果在 Kafka 中引入消费者确认机制，就需要 broker 维护消息消费状态，要做到高可靠就需要写文件持久化并与生产消息同步，这将急剧降低 Kafka 的性能，这种设计也极类似 RabbitMQ。如果不改变 Kafka 的实现，而是在 Kafka 和消费者之间做一层封装，还是需要实现一套类似 RabbitMQ 的消费确认和持久化机制。 参考资料：

这篇关于RabbitMQ 消息顺序、消息幂等、消息重复、消息事务、集群的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！