网站推广的方法有哪些百度刷排名seo软件

---

---

引言

每种MQ对顺序消息的实现都各有差异，我们今天主要来讲讲RocketMQ是如何实现顺序消息的。请注意，由于全局顺序消息不常用，我们这里提到的顺序消息都是指分区顺序消息。

顺序消息的定义😣

RocketMQ中的顺序消息分为全局顺序消息和分区顺序消息。

全局顺序消息

全局顺序消息是指某个Topic下的所有消息都能够按照发送的顺序被消费。这里稍微解释一下，为什么全局顺序消息不常用，因为全局顺序消息通常意味着在一个 Topic 下只有一个队列。如果要求整个 Topic 的所有消息都严格有序，那么必须确保所有的消息都发送到同一个队列，并且由同一个消费者实例按序消费。这就意味着，在这种情况下，该 Topic 只能配置一个队列。

然而，这也意味着性能受限于单个队列和单个消费者的处理能力，因为不能利用多队列带来的并行处理能力。因此，全局顺序消息适用于对顺序要求极高、但吞吐量需求不高的场景

所以大多数应用场景，更常见的是使用分区顺序消息，即保证某些类型的消息有序即可。

分区顺序消息

分区顺序消息则是指在消息被划分到不同的队列中时，每个队列内部的消息能够保证顺序消费。实际应用中，分区顺序消息更为常用，因为它能在保证一定顺序的同时提供更好的并发性能。

为了保证消息的顺序性，对于需要保证顺序的消息，生产者发送消息时必须确保相同业务逻辑的消息发送到同一个队列中。这通常是通过消息的Key进行哈希取模来决定消息发送到哪个队列。同时，消费者也需要采用单线程模式消费该队列中的消息，避免多线程环境下由于竞争导致的消息顺序错乱。

生产者端🤓

RocketMQ从生产者端和消费者端两个方面来保证了分区顺序消息机制，我们也从这两个方面来讲。

RocketMQ在生产者端通过MessageQueueSelector来保证顺序消息，传入orderId，id % mqs.size()保证消息会传入同一个队列，代表属于这个订单id（具体业务）的消息都传入这个队列，这里就可以保证队列局部有序

MessageQueueSelector 接口介绍

• mqs: 当前 Topic 对应的所有队列列表。
• msg: 当前要发送的消息对象。
• arg: 用户自定义参数，可以在发送消息时传入。
• 返回值：选择出的一个 MessageQueue，即这条消息将被发送到该队列。

public interface MessageQueueSelector {MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

使用MessageQueueSelector

下面是一个典型的使用 MessageQueueSelector 来保证订单消息顺序性的例子：

// 发送消息时指定 selector
String orderId = "ORDER_20250623_001";
producer.send(msg, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {Integer id = (Integer) arg;int index = id % mqs.size();return mqs.get(index);}
}, orderId.hashCode());

在这个例子中，所有具有相同 orderId 的消息都会被发送到同一个队列中，从而保证了这些消息在消费端是有序的。

消费者端😎

光有生产者用 MessageQueueSelector 把消息发到固定队列还不够，还需要消费者端配合才能保证顺序消费：

• 消费者必须采用 单线程消费模式 来消费每个队列的消息。
• RocketMQ 默认使用并发消费（多线程），所以需要设置 consumeThreadMin 和 consumeThreadMax 为 1，或者使用 MessageListenerOrderly 来监听消息。

ps：MessageListenerOrderly 保证的是：同一个队列中的消息，在一个消费者组内，只被同一个消费者实例串行消费

它确保每个队列的消息只能被该消费组中的一个消费者实例消费，并且是串行消费（即一条一条地处理，不并发），这样可以保证同一个队列中消息的顺序性。

那么同一个消费者在处理完这个一系列业务消息之前，不能处理其他队列的消息吗？虽然 MessageListenerOrderly 确保了单个 MessageQueue 内的消息是串行处理的，但它并不限制整个消费者实例同时处理多个不同的 MessageQueue。也就是说，不同 MessageQueue 上的消息仍然可以并行处理，只是在同一时间点上，每个 MessageQueue 只能由一个线程进行消费

示例：

consumer.registerMessageListener(new MessageListenerOrderly() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, MessageListenerOrderly.ConcurrentlyTopicSubSet context) {for (MessageExt msg : msgs) {System.out.println("消费消息: " + new String(msg.getBody()));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});

总结❤️

这就是RocketMQ实现顺序消息的解析

如果你看了这篇文章有收获可以点赞+关注+收藏🤩，这是对笔者更新的最大鼓励！如果你有更多方案或者文章中有错漏之处，请在评论区提出帮助笔者勘误，祝你拿到更好的offer！