我们知道,kafka 如果要保证顺序消费,必须保证消息保存到同一个patition上,而且为了有序性,只能有一个消费者进行消费。这种情况下,Kafka 就退化成了单一队列,毫无并发性可言,极大降低系统性能。那么对于对业务比较友好的RocketMQ 是如何实现的呢?首先,我们循序渐进的来了解下顺序消息的实现。
顺序消息业务使用场景
电商场景中传递订单状态。
同步mysql 的binlong 日志,数据库的操作是有顺序的。
其他消息之间有先后的依赖关系,后一条消息需要依赖于前一条消息的处理结果的情况。
等等。。。
消息中间件中的顺序消息
顺序消息(FIFO 消息)是 MQ 提供的一种严格按照顺序进行发布和消费的消息类型。顺序消息由两个部分组成:顺序发布和顺序消费。
顺序消息包含两种类型:
分区顺序:一个Partition(queue)内所有的消息按照先进先出的顺序进行发布和消费
全局顺序:一个Topic内所有的消息按照先进先出的顺序进行发布和消费.但是全局顺序极大的降低了系统的吞吐量,不符合mq的设计初衷。
那么折中的办法就是选择分区顺序。
【局部顺序消费】
如何保证顺序
在MQ的模型中,顺序需要由3个阶段去保障:
- 消息被发送时保持顺序
- 消息被存储时保持和发送的顺序一致
- 消息被消费时保持和存储的顺序一致
发送时保持顺序意味着对于有顺序要求的消息,用户应该在同一个线程中采用同步的方式发送。存储保持和发送的顺序一致则要求在同一线程中被发送出来的消息A和B,存储时在空间上A一定在B之前。而消费保持和存储一致则要求消息A、B到达Consumer之后必须按照先A后B的顺序被处理。
第一点,消息顺序发送,多线程发送的消息无法保证有序性,因此,需要业务方在发送时,针对同一个业务编号(如同一笔订单)的消息需要保证在一个线程内顺序发送,在上一个消息发送成功后,在进行下一个消息的发送。对应到mq中,消息发送方法就得使用同步发送,异步发送无法保证顺序性。
第二点,消息顺序存储,mq的topic下会存在多个queue,要保证消息的顺序存储,同一个业务编号的消息需要被发送到一个queue中。对应到mq中,需要使用MessageQueueSelector来选择要发送的queue,即对业务编号进行hash,然后根据队列数量对hash值取余,将消息发送到一个queue中。
第三点,消息顺序消费,要保证消息顺序消费,同一个queue就只能被一个消费者所消费,因此对broker中消费队列加锁是无法避免的。同一时刻,一个消费队列只能被一个消费者消费,消费者内部,也只能有一个消费线程来消费该队列。即,同一时刻,一个消费队列只能被一个消费者中的一个线程消费。
RocketMQ中顺序的实现
【Producer端】
Producer端确保消息顺序唯一要做的事情就是将消息路由到特定的分区,在RocketMQ中,通过MessageQueueSelector来实现分区的选择。
-
- public interface MessageQueueSelector {
-
-
- MessageQueue select(final List
mqs, final Message msg, final Object arg); - }
- List
mqs:消息要发送的Topic下所有的分区 - Message msg:消息对象
- 额外的参数:用户可以传递自己的参数
比如如下实现就可以保证相同的订单的消息被路由到相同的分区:
- long orderId = ((Order) object).getOrderId;
- return mqs.get(orderId % mqs.size());
【Consumer端】
尝试锁定锁定MessageQueue。
首先我们如何保证一个队列只被一个消费者消费?
消费队列存在于broker端,如果想保证一个队列被一个消费者消费,那么消费者在进行消息拉取消费时就必须向mq服务器申请队列锁,消费者申请队列锁的代码存在于RebalanceService消息队列负载的实现代码中。
消费者重新负载,并且分配完消费队列后,需要向mq服务器发起消息拉取请求,代码实现在RebalanceImpl#updateProcessQueueTableInRebalance中,针对顺序消息的消息拉取,mq做了如下判断:
- // 增加 不在processQueueTable && 存在于mqSet 里的消息队列。
- List
pullRequestList = new ArrayList<>(); // 拉消息请求数组 - for (MessageQueue mq : mqSet) {
- if (!this.processQueueTable.containsKey(mq)) {
- if (isOrder && !this.lock(mq)) { // 顺序消息锁定消息队列
- log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
- continue;
- }
-
- this.removeDirtyOffset(mq);
- ProcessQueue pq = new ProcessQueue();
- long nextOffset = this.computePullFromWhere(mq);
- if (nextOffset >= 0) {
- ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
- if (pre != null) {
- log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
- } else {
- log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
- PullRequest pullRequest = new PullRequest();
- pullRequest.setConsumerGroup(consumerGroup);
- pullRequest.setNextOffset(nextOffset);
- pullRequest.setMessageQueue(mq);
- pullRequest.setProcessQueue(pq);
- pullRequestList.add(pullRequest);
- changed = true;
- }
- } else {
- log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
- }
- }
- }
-
- // 发起消息拉取请求
- this.dispatchPullRequest(pullRequestList);
核心思想就是,消费客户端先向broker端发起对messageQueue的加锁请求,只有加锁成功时才创建pullRequest进行消息拉取,下面看下lock加锁请求方法:
-
- public boolean lock(final MessageQueue mq) {
- FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);
- if (findBrokerResult != null) {
- LockBatchRequestBody requestBody = new LockBatchRequestBody();
- requestBody.setConsumerGroup(this.consumerGroup);
- requestBody.setClientId(this.mQClientFactory.getClientId());
- requestBody.getMqSet().add(mq);
-
- try {
- // 请求Broker获得指定消息队列的分布式锁
- Set
lockedMq = - this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);
-
- // 设置消息处理队列锁定成功。锁定消息队列成功,可能本地没有消息处理队列,设置锁定成功会在lockAll()方法。
- for (MessageQueue mmqq : lockedMq) {
- ProcessQueue processQueue = this.processQueueTable.get(mmqq);
- if (processQueue != null) {
- processQueue.setLocked(true);
- processQueue.setLastLockTimestamp(System.currentTimeMillis());
- }
- }
-
- boolean lockOK = lockedMq.contains(mq);
- log.info("the message queue lock {}, {} {}",
- lockOK ? "OK" : "Failed",
- this.consumerGroup,
- mq);
- return lockOK;
- } catch (Exception e) {
- log.error("lockBatchMQ exception, " + mq, e);
- }
- }
-
- return false;
- }
代码实现逻辑比较清晰,就是调用lockBatchMQ方法发送了一个加锁请求,那么broker端收到加锁请求后的处理逻辑又是怎么样?
【broker端实现】
broker端收到加锁请求的处理逻辑在RebalanceLockManager#tryLockBatch方法中,RebalanceLockManager中关键属性如下:
-
- private final static long REBALANCE_LOCK_MAX_LIVE_TIME = Long.parseLong(System.getProperty(
- "rocketmq.broker.rebalance.lockMaxLiveTime", "60000"));
-
- private final Lock lock = new ReentrantLock();
-
- private final ConcurrentHashMap
> mqLockTable = - new ConcurrentHashMap<>(1024);
LockEntry对象中关键属性如下:
-
- static class LockEntry {
-
- private String clientId;
-
- private volatile long lastUpdateTimestamp = System.currentTimeMillis();
-
- public String getClientId() {
- return clientId;
- }
-
- public void setClientId(String clientId) {
- this.clientId = clientId;
- }
-
- public long getLastUpdateTimestamp() {
- return lastUpdateTimestamp;
- }
-
- public void setLastUpdateTimestamp(long lastUpdateTimestamp) {
- this.lastUpdateTimestamp = lastUpdateTimestamp;
- }
-
-
- public boolean isLocked(final String clientId) {
- boolean eq = this.clientId.equals(clientId);
- return eq && !this.isExpired();
- }
-
-
- public boolean isExpired() {
- boolean expired =
- (System.currentTimeMillis() - this.lastUpdateTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;
-
- return expired;
- }
- }
broker端通过对ConcurrentMap
【再次回到Consumer端,拿到锁后】
消费者对messageQueue的加锁已经成功,那么就进入到了第二个步骤,创建pullRequest进行消息拉取,消息拉取部分的代码实现在PullMessageService中,消息拉取完后,需要提交到ConsumeMessageService中进行消费,顺序消费的实现为ConsumeMessageOrderlyService,提交消息进行消费的方法为ConsumeMessageOrderlyService#submitConsumeRequest,具体实现如下:
- @Override
- public void submitConsumeRequest(//
- final List
msgs, // - final ProcessQueue processQueue, //
- final MessageQueue messageQueue, //
- final boolean dispathToConsume) {
- if (dispathToConsume) {
- ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
- this.consumeExecutor.submit(consumeRequest);
- }
- }
构建了一个ConsumeRequest对象,并提交给了ThreadPoolExecutor来并行消费,看下顺序消费的ConsumeRequest的run方法实现:
- public void run() {
- if (this.processQueue.isDropped()) {
- log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
- return;
- }
-
- // 获得 Consumer 消息队列锁
- final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
- synchronized (objLock) {
- // (广播模式) 或者 (集群模式 && Broker消息队列锁有效)
- if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
- || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
- final long beginTime = System.currentTimeMillis();
- // 循环
- for (boolean continueConsume = true; continueConsume; ) {
- if (this.processQueue.isDropped()) {
- log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
- break;
- }
-
- // 消息队列分布式锁未锁定,提交延迟获得锁并消费请求
- if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
- && !this.processQueue.isLocked()) {
- log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
- ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
- break;
- }
- // 消息队列分布式锁已经过期,提交延迟获得锁并消费请求
- if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
- && this.processQueue.isLockExpired()) {
- log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
- ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
- break;
- }
-
- // 当前周期消费时间超过连续时长,默认:60s,提交延迟消费请求。默认情况下,每消费1分钟休息10ms。
- long interval = System.currentTimeMillis() - beginTime;
- if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
- ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
- break;
- }
-
- // 获取消费消息。此处和并发消息请求不同,并发消息请求已经带了消费哪些消息。
- final int consumeBatchSize = ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
- List
msgs = this.processQueue.takeMessags(consumeBatchSize); - if (!msgs.isEmpty()) {
- final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);
-
- ConsumeOrderlyStatus status = null;
-
- // Hook:before
- ConsumeMessageContext consumeMessageContext = null;
- if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
- consumeMessageContext = new ConsumeMessageContext();
- consumeMessageContext
- .setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup());
- consumeMessageContext.setMq(messageQueue);
- consumeMessageContext.setMsgList(msgs);
- consumeMessageContext.setSuccess(false);
- // init the consume context type
- consumeMessageContext.setProps(new HashMap
()); - ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
- }
-
- // 执行消费
- long beginTimestamp = System.currentTimeMillis();
- ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
- boolean hasException = false;
- try {
- this.processQueue.getLockConsume().lock(); // 锁定队列消费锁
-
- if (this.processQueue.isDropped()) {
- log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
- this.messageQueue);
- break;
- }
-
- status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
- } catch (Throwable e) {
- log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}", //
- RemotingHelper.exceptionSimpleDesc(e), //
- ConsumeMessageOrderlyService.this.consumerGroup, //
- msgs, //
- messageQueue);
- hasException = true;
- } finally {
- this.processQueue.getLockConsume().unlock(); // 锁定队列消费锁
- }
-
- if (null == status //
- || ConsumeOrderlyStatus.ROLLBACK == status//
- || ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
- log.warn("consumeMessage Orderly return not OK, Group: {} Msgs: {} MQ: {}", //
- ConsumeMessageOrderlyService.this.consumerGroup, //
- msgs, //
- messageQueue);
- }
-
- // 解析消费结果状态
- long consumeRT = System.currentTimeMillis() - beginTimestamp;
- if (null == status) {
- if (hasException) {
- returnType = ConsumeReturnType.EXCEPTION;
- } else {
- returnType = ConsumeReturnType.RETURNNULL;
- }
- } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
- returnType = ConsumeReturnType.TIME_OUT;
- } else if (ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
- returnType = ConsumeReturnType.FAILED;
- } else if (ConsumeOrderlyStatus.SUCCESS == status) {
- returnType = ConsumeReturnType.SUCCESS;
- }
-
- if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
- consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
- }
-
- if (null == status) {
- status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
- }
-
- // Hook:after
- if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
- consumeMessageContext.setStatus(status.toString());
- consumeMessageContext
- .setSuccess(ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status);
- ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
- }
-
- ConsumeMessageOrderlyService.this.getConsumerStatsManager()
- .incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);
-
- // 处理消费结果
- continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
- } else {
- continueConsume = false;
- }
- }
- } else {
- if (this.processQueue.isDropped()) {
- log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
- return;
- }
-
- ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
- }
- }
- }
获取到锁对象后,使用synchronized尝试申请线程级独占锁。
如果加锁成功,同一时刻只有一个线程进行消息消费。
如果加锁失败,会延迟100ms重新尝试向broker端申请锁定messageQueue,锁定成功后重新提交消费请求
至此,第三个关键点的解决思路也清晰了,基本上就两个步骤。
创建消息拉取任务时,消息客户端向broker端申请锁定MessageQueue,使得一个MessageQueue同一个时刻只能被一个消费客户端消费。
消息消费时,多线程针对同一个消息队列的消费先尝试使用synchronized申请独占锁,加锁成功才能进行消费,使得一个MessageQueue同一个时刻只能被一个消费客户端中一个线程消费。
【顺序消费问题拆解】
- broke 上要保证一个队列只有一个进程消费,即一个队列同一时间只有一个consumer 消费
- broker 给consumer 的消息顺序应该保持一致,这个通过 rpc传输,序列化后消息顺序不变,所以很容易实现
- consumer 上的队列消息要保证同一个时间只有一个线程消费
通过问题的拆分,问题变成同一个共享资源串行处理了,要解决这个问题,通常的做法都是访问资源的时候加锁,即broker 上一个队列消息在被consumer 访问的必须加锁,单个consumer 端多线程并发处理消息的时候需要加锁;这里还需要考虑broker 锁的异常情况,假如一个broke 队列上的消息被consumer 锁住了,万一consumer 崩溃了,这个锁就释放不了,所以broker 上的锁需要加上锁的过期时间。
实际上 RocketMQ 消费端也就是照着上面的思路做:
RocketMQ中顺序消息注意事项
实际项目中并不是所有情况都需要用到顺序消息,但这也是设计方案的时候容易忽略的一点
顺序消息是生产者和消费者配合协调作用的结果,但是消费端保证顺序消费,是保证不了顺序消息的
消费端并行方式消费,只设置一次拉取消息的数量为 1(即配置参数 consumeBatchSize ),是否可以实现顺序消费 ?这里实际是不能的,并发消费在消费端有多个线程同时消费,consumeBatchSize 只是一个线程一次拉取消息的数量,对顺序消费没有意义,这里大家有兴趣可以看 ConsumeMessageConcurrentlyService 的代码,并发消费的逻辑都在哪里。
在使用顺序消息时,一定要注意其异常情况的出现,对于顺序消息,当消费者消费消息失败后,消息队列 RocketMQ 版会自动不断地进行消息重试(每次间隔时间为 1 秒),重试最大值是Integer.MAX_VALUE.这时,应用会出现消息消费被阻塞的情况。因此,建议您使用顺序消息时,务必保证应用能够及时监控并处理消费失败的情况,避免阻塞现象的发生。
重要的事再强调一次:在使用顺序消息时,一定要注意其异常情况的出现!防止资源不释放!
小结
通过以上的了解,我们知道了实现顺序消息所必要的条件:顺序发送、顺序存储、顺序消费。RocketMQ的设计中考虑到了这些,我们只需要简单的使用API,不需要额外使用代码来约束业务,使得实现顺序消息更加简单。