◆ 概述
在上文中，我们讨论了消费者对于消息拉取的实现，对于
这个黑盒的心脏部分，我们顺着消息的发送流程已经将其剖析了大半部分。本章我们不妨乘胜追击，接着讨论各种不同的消息的原理与实现。
◆ 事务消息◆ 概念
RocketMQ 中的事务消息功能，实际上是 分布式事务中的本地事务表 的实现，只不过，在这里用消息中间件来代替了数据库，同时也帮我们做好了回查的操作。
在这点上，RocketMQ 和 Kafka 是截然不同的，kafka 的事务是用来实现 Exacltly Once 语义，且该语义主要用来流计算中，即在 "从 Topic 中读 -> 计算 -> 存到 Topic" 保证不被重复计算。◆ 事务流程
客户端发送 half 消息
吐槽一下为什么要叫半消息(half message)，叫 prepare 消息不是更直观吗
Broker 将 half 消息持久化
客户端根据事务执行结果，发送 Commit / Rollback 消息
Broker 收到 Commit 时，将事务消息对消费者可见。收到 Rollback 时，将消息丢弃
Broker 过久未收到事务执行结果，询问客户端执行结果
客户端收到结果查询请求，执行回查方法，发送 Commit / Rollback 方法
Broker 根据事务执行结果做出对应处理
在设置好了事务监听器后（执行事务 与 事务回查），就可以发送事务消息
在将事务消息交给发送方法后，客户端首先会为消息添加事务消息的标识
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
然后将该事务消息会像普通的同步消息一样发送（且是同步发送）
sendResult = this.send(msg);
具体发送流程见：RocketMQ源码详解 | Producer篇 · 其一：Start，然后 Send 一条消息◆ 第二步
在 Broker 端接收到消息以后，会走与普通消息相同的底层通道（因为这个消息本身就只是个加上了 事务flag 的普通消息），然后由
TransactionalMessageService 来对这个消息进行额外处理。
首先会对该消息放入 real topic 属性和 real queue 属性，然后将消息 Topic 替换为用于处理所有事务消息的特殊的 Topic，当然该 Topic 对消费者是不可见的。
private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) { MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic()); MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msgInner.getQueueId())); // 设置标记为未收到结果 msgInner.setSysFlag( MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE)); // 替换到特殊的 Topic (RMQ_SYS_TRANS_HALF_TOPIC) msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic()); msgInner.setQueueId(0); msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); return msgInner;}
完成后，会送到 MessageStore 像普通消息一样处理
普通消息的具体流程见 RocketMQ源码详解 | Broker篇 · 其二：文件系统◆ 第三步
回到 Producer 端，在事务消息发送完成后，该方法会使用专门的线程池执行事务
// 2.执行本地事务，更新事务获取状态localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
然后对本地的事务执行状态进行处理，也就是将该执行状态上报
this.endTransaction(msg, sendResult, localTransactionState, localException);
这里会发送一条 oneway 命令给 Broker 端，且使用的是
RequestCode.END_TRANSACTION 请求码
// 事务结果报告(可能是 commit 或 rollback)public static final int END_TRANSACTION = 37;
完成处理后，该方法会将事务的发送结果和本地事务的执行结构都返回给上层 API
◆ 第四步
在 Broker 端，这里会由 EndTransactionProcessor 处理器来处理该请求码
然后，根据事务的执行结果来做不同的处理
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) { // 事务执行成功，尝试完成事务 // 获取 half 消息 result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader); if (result.getResponseCode() == ResponseCode.SUCCESS) { if (res.getCode() == ResponseCode.SUCCESS) { // 将 half 消息取出，构造真实消息，然后投入实际上的 Topic /* pass */ RemotingCommand sendResult = sendFinalMessage(msgInner); if (sendResult.getCode() == ResponseCode.SUCCESS) { /* * 找到半消息，进行删除 * 删除并不是物理上的删除,因为物理上的删除的代价十分的高昂,而是写入一条具有相同事务id的消息到 op Topic */ this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage()); } return sendResult; } return res; }}
如果需要回滚，则对相应的半消息进行删除，且和上面一样，并不是物理上的删除，而是发送具有相同事务 id 的消息到 OP Topic，来标记这个事务已经完成了(Commit/Rollback)， OP Topic 也是一个特殊的 Topic，同样对消费者不可见。
if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) { // 事务执行失败，进行 half 消息的回滚 // 首先找到 half 消息 result = this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader); if (result.getResponseCode() == ResponseCode.SUCCESS) { RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader); if (res.getCode() == ResponseCode.SUCCESS) { // 进行删除 this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage()); } return res; }}
当这些都做完后，一次事务就完成了。
◆ 补偿
当然啦，以上是顺利的情况，我们当然不能指望事务每一次都能执行成功、网络分区和宕机事件永远不会发生。
在一段时间后，如果客户端没有对事务的状态进行上报（或者上报的状态不是 Commit 或 Rollback，而是 Unknown）， Broker 端当然就要进行事务状态的回查。
在 BrokerController 启动的时候，会开启事务状态检测服务，该服务会通过循环调用
TransactionalMessageServiceImpl.check() 方法，不断的扫描未结束的事务，同时对超过指定时间还不知道状态的事务进行回查操作。
check() 方法是事务回查的核心，由于很长，我们先来看第一部分(删减了没人在意的 Log)
// 首先找到存储所有 half 消息的 TopicString topic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;Set msgQueues = transactionalMessageBridge.fetchMessageQueues(topic);/ / 对其中每一个 queue 进行检查for (MessageQueue messageQueue : msgQueues) { long startTime = System.currentTimeMillis(); // 获得对应的 op 消息所在的 queue MessageQueue opQueue = getOpQueue(messageQueue); // 获取未处理的 half 消息的起始偏移量 long halfOffset = transactionalMessageBridge.fetchConsumeOffset(messageQueue); // 获取 op 消息的 queue 的起始偏移量 long opOffset = transactionalMessageBridge.fetchConsumeOffset(opQueue); // 用来记录已经被处理了的 op 消息的偏移量 List doneOpOffset = new ArrayList<>(); / / 用来记录已经完成了的 half 消息的偏移量 // key: halfOffset, value: opOffset HashMap removeMap = new HashMap<>(); PullResult pullResult = fillOpRemoveMap(removeMap, opQueue, opOffset, halfOffset, doneOpOffset);
在 fillOpRemoveMap 方法中，主要是将 op 消息取出，来标记可以被移除的 half 消息（op 消息的存在代表对应事务的结束）
/** * 读取op消息，解析op消息，填充removeMap * * @param removeMap 要删除的半消息，key: halfOffset，value: opOffset * @param opQueue Op message queue. * @param pullOffsetOfOp op message queue 的起始偏移量 * @param miniOffset half message queue 的当前最小偏移量 * @param doneOpOffset 存储已处理的 op 消息 * @return 获取到的 Op 消息 */private PullResult fillOpRemoveMap(HashMap removeMap, MessageQueue opQueue, long pullOffsetOfOp, long miniOffset, List doneOpOffset) { / / 首先通过 queue 获取 op 消息，最大数量为 32 条 PullResult pullResult = pullOpMsg(opQueue, pullOffsetOfOp, 32); /* pass: pullResult 消息的意外状态的处理 */ List opMsg = pullResult.getMsgFoundList(); for (MessageExt opMessageExt : opMsg) { / / op 消息的 body 存储的是对应的 half 消息的偏移量, 现在将其取出 Long queueOffset = getLong(new String(opMessageExt.getBody(), TransactionalMessageUtil.charset)); // 感觉这里的 Tag 并没有什么意义，无论是 Commit 还是 Rollback 都会加入这个 Tag if (TransactionalMessageUtil.REMOVETAG.equals(opMessageExt.getTags())) { // 在 已处理偏移量 之前的话则可直接放入 已处理偏移量集合 if (queueOffset < miniOffset) { doneOpOffset.add(opMessageExt.getQueueOffset()); } else { // 否则放入需要移除的 half 的消息的集合 removeMap.put(queueOffset, opMessageExt.getQueueOffset()); } } } return pullResult;}
然后进入到 check 方法的第二部分
while (true) { if (System.currentTimeMillis() - startTime > MAX_PROCESS_TIME_LIMIT) break; // 推进最小已处理偏移量 if (removeMap.containsKey(i)) /* 如果该 half 消息存在对应的 op 消息，说明已经被处理了(commit/rollback) */ { // 取出放入到已处理偏移量队列 Long removedOpOffset = removeMap.remove(i); doneOpOffset.add(removedOpOffset); } else /* 否则说明当前 half 消息悬而未决 */ { // 取出对应的半消息 GetResult getResult = getHalfMsg(messageQueue, i); /* pass: 半消息不存在时的意外处理 */ /* * 检测是否要丢弃或跳过 * 丢弃条件: 当前事务已经超过了最大回查次数(15次) * 跳过条件: 已经超过了过期文件最大保留时间(72小时) */ if (needDiscard(msgExt, transactionCheckMax) || needSkip(msgExt)) { // 处理并推进偏移量 // 具体的处理方法是: 投入 TRANS_CHECK_MAX_TIME_TOPIC 这个 Topic，等待手动处理 listener.resolveDiscardMsg(msgExt); // 进入到下一个 half 消息 newOffset = i + 1; i++; continue; } if (msgExt.getStoreTimestamp() >= startTime) { break; }
上面的方法很好理解，只是对于已经被标记结束的事务的处理、和未结束事务的补足
接下来是第三部分，这里将继续对未结束事务的补足，与进行可能的回查操作
// half 消息具有最小的检查时间(免疫时间), 检测时间以内可以跳过回查, 重新投入 half 消息的 Topic long valueOfCurrentMinusBorn = System.currentTimeMillis() - msgExt.getBornTimestamp(); long checkImmunityTime = transactionTimeout; String checkImmunityTimeStr = msgExt.getUserProperty(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS); if (null != checkImmunityTimeStr) { checkImmunityTime = getImmunityTime(checkImmunityTimeStr, transactionTimeout); if (valueOfCurrentMinusBorn < checkImmunityTime) { if (checkPrepareQueueOffset(removeMap, doneOpOffset, msgExt)) { newOffset = i + 1; i++; continue; } } } else { if ((0 <= valueOfCurrentMinusBorn) && (valueOfCurrentMinusBorn < checkImmunityTime)) { break; } } /* * 对于当前事务的回查操作，需要满足三个条件之一 * 1.当前 op 消息的集合为空，且已经超过了最小检查时间(免疫时间) * 2.最大偏移量的 op 消息的生成时间 已经超过了 最小检查时间 * 3.关闭最小检查时间 */ List opMsg = pullResult.getMsgFoundList(); boolean isNeedCheck = (opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime) || (opMsg != null && (opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout)) || (valueOfCurrentMinusBorn <= -1); if (isNeedCheck) { // 先将当前 half 消息放回 if (!putBackHalfMsgQueue(msgExt, i)) { continue; } // 然后向 Product 发送检测消息 listener.resolveHalfMsg(msgExt); } else { // 否则更新 op 消息集合,以确保能够断言该 half 消息的状态 pullResult = fillOpRemoveMap(removeMap, opQueue, pullResult.getNextBeginOffset(), halfOffset, doneOpOffset); continue; }} newOffset = i + 1; i++;}
上面这段代码主要围绕 "是否进行回查" 展开，且涉及到 "免疫时间"。
在一个事务消息被发送后，对应事务的执行当然需要一定的执行时间，如果我们不设置这个时间立刻进行回查，那么很有可能时候事务还没执行完，对于大多数情况下还没执行完的事务进行回查，毫无疑问带来的收益很低。所以我们需要设定一个时间，在这个时间内的事务先暂时不回查，这个时间就叫做"免疫时间"。
然后再来看下需要进行回查的三种情况：
当 op 消息的集合为空，说明当前还没有收到让当前事务结束的通知，且超过了"免疫时间"，故回查
当前 op 消息最大偏移量的生成时间超过了"免疫时间"，说明该事务的提交消息可能丢失了，故回查
不启用 "免疫时间"
其中发送的回查消息的请求码为
RequestCode.CHECK_TRANSACTION_STATE ，发送的也是 oneway 消息
最后的第四部分，同时更新 half 和 op 消息在 Queue 中的偏移量
// 对所有的 half 消息计算完成后，更新偏移量if (newOffset != halfOffset) { transactionalMessageBridge.updateConsumeOffset(messageQueue, newOffset);}// 根据已经被标记为完成的 op 消息更新偏移量long newOpOffset = calculateOpOffset(doneOpOffset, opOffset);if (newOpOffset != opOffset) { // 如果不等，说明并不是所有的 op 消息都被标记为完成了 // 所以我们只将偏移量更新到第一个未完成的 op 消息的位置，其后面的 op 消息会在下次重复处理 transactionalMessageBridge.updateConsumeOffset(opQueue, newOpOffset);}
然后在 Producer 这边，将由
ClientRemotingProcessor.checkTransactionState() 来处理回查操作
// 获取事务 IDString transactionId = messageExt.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);if (null != transactionId && !"".equals(transactionId)) { messageExt.setTransactionId(transactionId);}final String group = messageExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);if (group != null) { // 从 MQClientFactory 找到注册的对应 Producer MQProducerInner producer = this.mqClientFactory.selectProducer(group); if (producer != null) { final String addr = RemotingHelper.parseChannelRemoteAddr(ctx.channel()); // 让 Producer 检查在对应 IP 上的事务状态 producer.checkTransactionState(addr, messageExt, requestHeader); } else { log.debug("checkTransactionState, pick producer by group[{}] failed", group); }} else { log.warn("checkTransactionState, pick producer group failed");}
再进入
producer.checkTransactionState() 看看 Producer 是怎样检查事务状态的
TransactionCheckListener transactionCheckListener = DefaultMQProducerImpl.this.checkListener();// 取出当前 Producer 的事务监听器TransactionListener transactionListener = getCheckListener();if (transactionCheckListener != null || transactionListener != null) { LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW; Throwable exception = null; try { if (transactionCheckListener != null) { // 调用其的事务回查方法 localTransactionState = transactionCheckListener.checkLocalTransactionState(message); } else if (transactionListener != null) { log.debug("Used new check API in transaction message"); localTransactionState = transactionListener.checkLocalTransaction(message); } } catch (Throwable e) { log.error("Broker call checkTransactionState, but checkLocalTransactionState exception", e); exception = e; } // 再将事务执行结果其发回给 Broker this.processTransactionState( localTransactionState, group, exception);} else { log.warn("CheckTransactionState, pick transactionCheckListener by group[{}] failed", group);}
最后发回的方法做的事情和在一开始发送事务状态的方法，所做的事情是一样的。Broker 做的处理也是一样的。
这样，补偿流程就执行完了。
◆ 批量消息◆ 概念
在消息队列中，批量消息也是一个重要的部分，将消息压缩在一起发送不仅可以减少带宽的消耗，还能节省头部占用的空间。
有点失望的是，RocketMQ 对于批量消息的实现有点"粗糙"了◆ 源码流程
首先，在调用 send() 的 batch 版本后，会先对批量消息进行校验
批量消息不允许延时、不允许发送到重试 Topic，且要求发送到的 Topic 必须是同一个 Topic
List messageList = new ArrayList (messages.size());Message first = null; for (Message message : messages) { if (message.getDelayTimeLevel() > 0) { throw new UnsupportedOperationException( "TimeDelayLevel is not supported for batching"); } if (message.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { throw new UnsupportedOperationException( "Retry Group is not supported for batching"); } if (first == null) { first = message; } else { if (!first.getTopic(). equals(message.getTopic())) { throw new UnsupportedOperationException( "The topic of the messages in one batch should be the same"); } if (first.isWaitStoreMsgOK() != message.isWaitStoreMsgOK()) { throw new UnsupportedOperationException( "The waitStoreMsgOK of the messages in one batch should the same"); } } messageList. add(message);}MessageBatch messageBatch = new MessageBatch(messageList);
在校验完成，且都放到一个 List 之后，接下来的步骤和普通的消息发送都差不多，只是在编码上理所当然的存在着不同
public static byte[] encodeMessages(List messages) { //TO DO refactor, accumulate in one buffer, avoid copies List< byte[]> encodedMessages = new ArrayList< byte[]>(messages.size()); int allSize = 0; for (Message message : messages) { // 编码每一个消息 byte[] tmp = encodeMessage(message); encodedMessages.add(tmp); allSize += tmp.length; } // 放到最后的大集合中 byte[] allBytes = new byte[allSize]; int pos = 0; for ( byte[] bytes : encodedMessages) { System.arraycopy(bytes, 0, allBytes, pos, bytes.length); pos += bytes.length; } return allBytes;}
然后使用
RequestCode.SEND_BATCH_MESSAGE 这个状态码发送出去。
在 Broker 端，其投入的过程大体上和普通消息类似，但是其最后的持久化到硬盘时，这块批量消息被拆分为了普通的单条消息。
即 RocketMQ 使用批量消息只减少了发送时的宽带传输，对于存储与交给消费者的部分并没有获得优化
// 拆分批量消息为每一个普通消息while (messagesByteBuff.hasRemaining()) { // 1 TOTALSIZE final int msgPos = messagesByteBuff.position(); final int msgLen = messagesByteBuff.getInt(); final int bodyLen = msgLen - 40; //only for log, just estimate it /* pass: 当作普通消息存储 */ queueOffset++; msgNum++; messagesByteBuff.position(msgPos + msgLen);}
◆ 延时消息◆ 概念
在业务中，有时候有一些延时提交任务的需求，这时候就可以使用延时消息，即在投递一部分时间后才对消费者可见。
不过，在 RocketMQ 中，延迟级别并不支持自定义，而是具有固定的延迟级别。
不过商业版的 阿里云MQ 可以支持秒精度的自定义延迟时间，果然是为了阉割社区版来赚钱吗◆ 源码流程
RocketMQ 对于延时消息的处理主要在于 Broker 端，所以我们只需要看在 Broker 对延时级别的处理。
首先，在 CommitLog 的 put 中，会对延迟级别进行判断，如果存在，会在这进行进行 Topic 的替换，将其存储到对应的延迟级别的 Queue
if (msg.getDelayTimeLevel() > 0) { if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) { msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()); } topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC; queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel()); // Backup real topic, queueId MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic()); MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId())); msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties())); msg.setTopic(topic); msg.setQueueId(queueId);}
然后会被在 DefaultMessageStore 中初始化的 ScheduleMessageService 处理
首先，该服务在启动时会进行初始化
public void start() { // 保证只被执行一次 if (started.compareAndSet(false, true)) { // 加载本地快照 super.load(); this.timer = new Timer("ScheduleMessageTimerThread", true); for (Map.Entry entry : this.delayLevelTable.entrySet()) { // 取出每一个级别 Integer level = entry.getKey(); // 当前延迟级别对应的延迟时间 Long timeDelay = entry.getValue(); // 该延迟级别之前消费到的自己的队列的偏移量 Long offset = this.offsetTable.get(level); if ( null == offset) { offset = 0L; } // 每一个延迟级别设置一个定时任务 if (timeDelay != null) { this.timer.schedule( new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME); } } // 定时持久化各个延迟级别的偏移量 this.timer.scheduleAtFixedRate( new TimerTask() { @Override public void run() { try { if (started.get()) ScheduleMessageService. this.persist(); } catch (Throwable e) { log.error( "scheduleAtFixedRate flush exception", e); } } }, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval()); }}
每一个延迟级别的 Queue 都有对应的定时任务，且都会执行以下方法
public void executeOnTimeup() { // 找到自己延迟级别的消费队列 ConsumeQueue cq = ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel2QueueId(delayLevel)); long failScheduleOffset = offset; if (cq != null) { // 根据消费偏移量将指定的 MappedFile 文件加载进来 SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset); if (bufferCQ != null) { try { long nextOffset = offset; int i = 0; // 遍历每一个消息的索引 ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit(); for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) { long offsetPy = bufferCQ.getByteBuffer().getLong(); int sizePy = bufferCQ.getByteBuffer().getInt(); long tagsCode = bufferCQ.getByteBuffer().getLong(); /* pass */ long now = System.currentTimeMillis(); long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode); nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); long countdown = deliverTimestamp - now; if (countdown <= 0) /* 目标时间小于当起时间，可以执行 */ { // 根据偏移量取出消息 MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset( offsetPy, sizePy); if (msgExt != null) { try { // 将延迟消息恢复成原本消息的样子 MessageExtBrokerInner msgInner = this.messageTimeup(msgExt); /* pass */ // 投入真实的 Topic PutMessageResult putMessageResult = ScheduleMessageService.this.writeMessageStore .putMessage(msgInner); /* pass: 更新度量信息 */ } catch (Exception e) { /* pass */ } } } else /* 否则，这个消息需要被消费的时间到了再通知我 */ { ScheduleMessageService.this.timer.schedule( new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset), countdown); // 更新消费偏移量 ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset); return; } } // end of for // 走到这里，说明暂时没有需要消费的延时消息 nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); // 小睡一会 ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask( this.delayLevel, nextOffset), DELAY_FOR_A_WHILE); ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset); return; } finally { bufferCQ.release(); } } // end of if (bufferCQ != null) /* pass */ } // end of if (cq != null) ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel, failScheduleOffset), DELAY_FOR_A_WHILE);}
可以看出，延迟消息的实现还是十分简单的，由于先投入的延时消息必先快于后投入的消息的到期，所以只需要不断的拉取各个延迟级别对应的队列 的头部的延迟消息即可。这也是只支持固定级别的延迟消息带来的好处。
架构设计之架构演化、模式与核心要素
架构设计，如何业务逻辑和技术分离？
“12306” 经历过那些优化？
什么是架构及架构的本质？
架构杂谈
了解新兴架构模式：微应用架构
微服务架构设计方案详解
免责声明：
本公众号部分分享的资料来自网络收集和整理，所有文字和图片版权归属于原作者所有，且仅代表作者个人观点，与本公众号无关，文章仅供读者学习交流使用，并请自行核实相关内容，如文章内容涉及侵权，请联系后台管理员删除。
