您的位置:首页 > 科技 > IT业 > html音乐网页设计模板_大公司网页设计用什么软件好_企业网站模板html_网络推广合作资源平台

html音乐网页设计模板_大公司网页设计用什么软件好_企业网站模板html_网络推广合作资源平台

2025/3/18 9:22:33 来源:https://blog.csdn.net/m0_71845127/article/details/145990210  浏览:    关键词:html音乐网页设计模板_大公司网页设计用什么软件好_企业网站模板html_网络推广合作资源平台
html音乐网页设计模板_大公司网页设计用什么软件好_企业网站模板html_网络推广合作资源平台

RocketMQ的顺序消费机制通过生产端和消费端的协同设计实现,其核心在于局部顺序性,即保证同一队列(MessageQueue)内的消息严格按发送顺序消费。以下是详细机制解析及关键源码实现:
在这里插入图片描述


一、顺序消费的核心机制

1. 生产端路由策略
  • Sharding Key路由:生产者通过MessageQueueSelector接口将同一业务标识(如订单ID)的消息路由到同一队列。例如,根据订单ID对队列数取模,确保同一订单的消息进入同一队列。
    // 示例:生产者选择队列
    SendResult sendResult = 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);
    

路由方法:
在这里插入图片描述
SelectMessageQueueByHash:按哈希选择消息队列。
SelectMessageQueueByRandom:随机选择消息队列。
SelectMessageQueueByMachineRoom:按照机房选择消息队列。

  • 同步发送:必须使用同步发送(send()方法),异步发送无法保证消息顺序。
2. 消费端锁机制
  • Broker端队列锁:消费者集群模式下,通过定时任务(默认每20秒)向Broker申请队列锁,只有获得锁的消费者实例才能拉取并消费该队列消息。锁的有效期默认60秒,避免宕机导致死锁。
  • 本地队列快照锁:消费者在消费时对ProcessQueue(队列快照)加内存锁(synchronized块),确保同一队列的消息仅由一个线程顺序处理。
3. 消费流程控制
  • 单线程顺序消费:每个队列对应一个消费线程,从ProcessQueue的红黑树(msgTreeMap)中按消息偏移量顺序取出消息,保证消费顺序与存储顺序一致。
  • 失败重试机制:消费失败时,若未达最大重试次数,消息会重新放回ProcessQueue等待下次消费;若超过次数则进入死信队列。

二、关键源码解析

1. 消费者启动与锁管理
  • 服务初始化:消费者启动时,若监听器为MessageListenerOrderly,则创建ConsumeMessageOrderlyService,并启动定时加锁任务。

    // DefaultMQPushConsumerImpl#start
    if (getMessageListenerInner() instanceof MessageListenerOrderly) {this.consumeMessageService = new ConsumeMessageOrderlyService(this, listener);consumeMessageService.start();
    }
    
  • 定时加锁ConsumeMessageOrderlyService启动后,定时调用RebalanceImpl.lockAll()向Broker申请锁,更新ProcessQueue的锁定状态。

      public synchronized void lockMQPeriodically() {if (!this.stopped) {this.defaultMQPushConsumerImpl.getRebalanceImpl().lockAll();}}
    
      				for (MessageQueue mq : mqs) {ProcessQueue processQueue = this.processQueueTable.get(mq);if (processQueue != null) {if (lockOKMQSet.contains(mq)) {if (!processQueue.isLocked()) {log.info("the message queue locked OK, Group: {} {}", this.consumerGroup, mq);}// 更新`ProcessQueue`的锁定状态 trueprocessQueue.setLocked(true);processQueue.setLastLockTimestamp(System.currentTimeMillis());} else {// 更新`ProcessQueue`的锁定状态 falseprocessQueue.setLocked(false);log.warn("the message queue locked Failed, Group: {} {}", this.consumerGroup, mq);}}}
    
2. 消息拉取与消费
  • 锁检查:拉取消息前检查ProcessQueue是否已锁定,未锁定则延迟拉取。
    // DefaultMQPushConsumerImpl#pullMessage
    if (processQueue.isLocked()) {// 计算消费偏移量并拉取消息
    } else {executePullRequestLater(pullRequest, 3000); // 延迟3秒重试
    }
    
  • 消费线程加锁:消费线程运行时获取队列内存锁,确保单线程处理。
    synchronized (messageQueueLock.fetchLockObject(messageQueue)) {List<MessageExt> msgs = processQueue.takeMessags(batchSize);// 执行消费逻辑
    }
    
3. Broker端锁管理
  • 锁存储:Broker通过RebalanceLockManager维护锁信息,记录消费者ClientID和最后更新时间,超时(默认60秒)则自动释放。
    class LockEntry {String clientId;long lastUpdateTimestamp;boolean isExpired() { /* 检查是否超时 */ }
    }
    
  • 锁竞争:消费者通过lockBatchMQ请求批量加锁,Broker返回成功锁定的队列列表。

三、适用场景与注意事项

  1. 适用场景

    • 分区顺序:如订单流程(创建、支付、完成),同一订单ID的消息需顺序处理。
    • 全局顺序Topic仅一个队列,性能较低,适用于强一致性场景(如证券交易)。
  2. 注意事项

    • 幂等性:因网络抖动或消费者重启可能导致短暂乱序,业务逻辑需支持幂等处理。
    • 队列数选择:分区数越多并发度越高,但需确保同一业务ID的路由一致性。

总结

RocketMQ的顺序消费通过生产端路由策略消费端锁机制Broker协同管理实现。其设计在保证局部顺序的同时兼顾性能,适用于多数业务场景。源码层面,ConsumeMessageOrderlyServiceRebalanceImpl是核心模块,通过定时加锁单线程消费队列快照管理确保顺序性。实际使用时需结合业务特点设计Sharding Key,并处理可能的异常情况。

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com