RocketMQ事务消息学习及刨坑过程

34次阅读

共计 17078 个字符，预计需要花费 43 分钟才能阅读完成。

一、背景

MQ 组件是系统架构里必不可少的一门利器，设计层面可以降低系统耦合度，高并发场景又可以起到削峰填谷的作用，从单体应用到集群部署方案，再到现在的微服务架构，MQ 凭借其优秀的性能和高可靠性，得到了广泛的认可。
随着数据量增多，系统压力变大，开始出现这种现象：数据库已经更新了，但消息没发出来，或者消息先发了，但后来数据库更新失败了，结果研发童鞋各种数据修复，这种生产问题出现的概率不大，但让人很郁闷。这个其实就是数据库事务与 MQ 消息的一致性问题，简单来讲，数据库的事务跟普通 MQ 消息发送无法直接绑定与数据库事务绑定在一起，例如上面提及的两种问题场景：

  1. 数据库事务提交后发送 MQ 消息；
  2. MQ 消息先发，然后再提交数据库事务。

场景 1 的问题是数据库事务可能刚刚提交，服务器就宕机了，MQ 消息没发出去，场景 2 的问题就是 MQ 消息发送出去了，但数据库事务提交失败，又没办法追加已经发出去的 MQ 消息，结果导致数据没更新，下游已经收到消息，最终事务出现不一致的情况。

二、事务消息的引出

我们以微服务架构的购物场景为例，参照一下 RocketMQ 官方的例子，用户 A 发起订单，支付 100 块钱操作完成后，能得到 100 积分，账户服务和会员服务是两个独立的微服务模块，有各自的数据库，按照上文提及的问题可能性，将会出现这些情况：

  • 如果先扣款，再发消息，可能钱刚扣完，宕机了，消息没发出去，结果积分没增加。
  • 如果先发消息，再扣款，可能积分增加了，但钱没扣掉，白送了 100 积分给人家。
  • 钱正常扣了，消息也发送成功了，但会员服务实例消费消息出现问题，结果积分没增加。

由此引出的是数据库事务与 MQ 消息的事务一致性问题，rocketmq 事务消息解决的问题：解决本地事务执行与消息发送的原子性问题。这里界限一定要明白，是确保 MQ 生产端正确无误地将消息发送出来，没有多发，也不会漏发。但至于发送后消费端有没有正常的消费掉（如上面提及的第三种情况，钱正常扣了，消息也发了，但下游消费出问题导致积分不对），这种异常场景将由 MQ 消息消费失败重试机制来保证，不在此次的讨论范围内。

常用的 MQ 组件针对此场景都有自己的实现方案，如 ActiveMQ 使用 AMQP 协议 (二阶提交方式) 保证消息正确发送，这里我们以 RocketMQ 为重点进行学习。

三、RocketMQ 事务消息设计思路

根据 CAP 理论，RocketMQ 事务消息通过异步确保方式，保证事务的最终一致性。设计流程上借鉴两阶段提交理论，流程图如下：

  1. 应用模块遇到要发送事务消息的场景时，先发送 prepare 消息给 MQ。
  2. prepare 消息发送成功后，应用模块执行数据库事务（本地事务）。
  3. 根据数据库事务执行的结果，再返回 Commit 或 Rollback 给 MQ。
  4. 如果是 Commit，MQ 把消息下发给 Consumer 端，如果是 Rollback，直接删掉 prepare 消息。
  5. 第 3 步的执行结果如果没响应，或是超时的，启动定时任务回查事务状态（最多重试 15 次，超过了默认丢弃此消息），处理结果同第 4 步。
  6. MQ 消费的成功机制由 MQ 自己保证。

四、RocketMQ 事务消息实现流程

以 RocketMQ 4.5.2 版本为例，事务消息有专门的一个队列 RMQ_SYS_TRANS_HALF_TOPIC，所有的 prepare 消息都先往这里放，当消息收到 Commit 请求后，就把消息再塞到真实的 Topic 队列里，供 Consumer 消费，同时向 RMQ_SYS_TRANS_OP_HALF_TOPIC 塞一条消息。简易流程图如下：

上述流程中，请允许我这样划分模块职责：

  1. RocketMQ Client 即我们工程中导入的依赖 jar 包，RocketMQ Broker 端即部署的服务端，NameServer 暂未体现。
  2. 应用模块成对出现，上游为事务消息生产端，下游为事务消息消费端（事务消息对消费端是透明的，与普通消息一致）。

应用模块的事务因为中断，或是其他的网络原因，导致无法立即响应的，RocketMQ 当做 UNKNOW 处理，RocketMQ 事务消息还提供了一个补救方案：定时查询事务消息的数据库事务状态
简易流程图如下：

五、源码剖析

讲解的思路基本上按照如下流程图，根据模块职责和流程逐一分析。

  1. 环境准备

阅读源码前需要在 IDE 上获取和调试 RocketMQ 的源码，这部分请自行查阅方法。

  1. 应用模块（事务消息生产端）核心源码

创建一个监听类，实现 TransactionListener 接口，在实现的数据库事务提交方法和回查事务状态方法模拟结果。

/**
 * @program: rocket
 * @description: 调试事务消息示例代码
 * @author: Huang
 * @create: 2019-10-16
 **/
public class SelfTransactionListener implements TransactionListener {private AtomicInteger transactionIndex = new AtomicInteger(0);
   private AtomicInteger checkTimes = new AtomicInteger(0);

   private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
   /**
    * 执行本地事务
    *
    * @param message
    * @param o
    * @return
    */
   @Override
   public LocalTransactionState executeLocalTransaction(Message message, Object o) {String msgKey = message.getKeys();
      System.out.println("start execute local transaction" + msgKey);
      LocalTransactionState state;
      if (msgKey.contains("1")) {
         // 第一条消息让他通过
         state = LocalTransactionState.COMMIT_MESSAGE;
      } else if (msgKey.contains("2")) {
         // 第二条消息模拟异常，明确回复回滚操作
         state = LocalTransactionState.ROLLBACK_MESSAGE;
      } else {
         // 第三条消息无响应，让它调用回查事务方法
         state = LocalTransactionState.UNKNOW;
         // 给剩下 3 条消息，放 1，2，3 三种状态
         localTrans.put(msgKey, transactionIndex.incrementAndGet());
      }
      System.out.println("executeLocalTransaction:" + message.getKeys() + ",execute state:" + state + ",current time：" + System.currentTimeMillis());
      return state;
   }

   /**
    * 回查本地事务结果
    *
    * @param messageExt
    * @return
    */
   @Override
   public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {String msgKey = messageExt.getKeys();
      System.out.println("start check local transaction" + msgKey);
      Integer state = localTrans.get(msgKey);
      switch (state) {
         case 1:
            System.out.println("check result unknown 回查次数" + checkTimes.incrementAndGet());
            return LocalTransactionState.UNKNOW;
         case 2:
            System.out.println("check result commit message, 回查次数" + checkTimes.incrementAndGet());
            return LocalTransactionState.COMMIT_MESSAGE;
         case 3:
            System.out.println("check result rollback message, 回查次数" + checkTimes.incrementAndGet());
            return LocalTransactionState.ROLLBACK_MESSAGE;

         default:
            return LocalTransactionState.COMMIT_MESSAGE;
      }
   }
}

事务消息生产者代码示例，共发送 5 条消息，基本上包含全部的场景，休眠时间设置足够的时间，保证回查事务时实例还在运行中，代码如下：

/**
 * @program: rocket
 * @description: Rocketmq 事务消息
 * @author: Huang
 * @create: 2019-10-16
 **/
public class TransactionProducer {public static void main(String[] args) {
      try {TransactionMQProducer producer = new TransactionMQProducer("transactionMQProducer");
         producer.setNamesrvAddr("10.0.133.29:9876");
         producer.setTransactionListener(new SelfTransactionListener());
         producer.start();
         for (int i = 1; i < 6; i++) {Message message = new Message("TransactionTopic", "transactionTest","msg-" + i, ("Hello" + ":" +  i).getBytes());
            try {SendResult result = producer.sendMessageInTransaction(message, "Hello" + ":" +  i);
               System.out.printf("Topic:%s send success, misId is:%s%n", message.getTopic(), result.getMsgId());
            } catch (Exception e) {e.printStackTrace();
            }
         }
         Thread.sleep(Integer.MAX_VALUE);
         producer.shutdown();} catch (MQClientException e) {e.printStackTrace();
      } catch (InterruptedException e) {e.printStackTrace();
      }
   }
}
  1. RocketMQ Client 端代码，代码主要逻辑可以分成三段：第一段为设置消息为 prepare 消息，并发送给 RocketMQ 服务端
SendResult sendResult = null;
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
try {sendResult = this.send(msg);
} catch (Exception e) {throw new MQClientException("send message Exception", e);
}

第二段：消息发送成功后，调用应用模块数据库事务方法，获取事务结果（为节省篇幅，代码有删节）

switch (sendResult.getSendStatus()) {
    case SEND_OK: {
        try {if (null != localTransactionExecuter) {localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
            } else if (transactionListener != null) {log.debug("Used new transaction API");
                localTransactionState = transactionListener.executeLocalTransaction(msg, arg);
            }
            if (null == localTransactionState) {localTransactionState = LocalTransactionState.UNKNOW;}
        } catch (Throwable e) {log.info("executeLocalTransactionBranch exception", e);
            log.info(msg.toString());
            localException = e;
        }
    }
    break;
    case FLUSH_DISK_TIMEOUT:
    case FLUSH_SLAVE_TIMEOUT:
    case SLAVE_NOT_AVAILABLE:
        localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
        break;
    default:
        break;
}

第三段：发送事务结果到 RocketMQ 端，结束事务，并响应结果给应用模块

try {this.endTransaction(sendResult, localTransactionState, localException);
} catch (Exception e) {log.warn("local transaction execute" + localTransactionState + ", but end broker transaction failed", e);
}
  1. RocketMQ Broker 端事务提交 / 回滚操作（这里取 endTransaction 部分）

代码入口：org.apache.rocketmq.broker.processor.EndTransactionProcessor

OperationResult result = new OperationResult();
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);
    if (result.getResponseCode() == ResponseCode.SUCCESS) {RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
        if (res.getCode() == ResponseCode.SUCCESS) {
            // 修改消息的 Topic 为由 RMQ_SYS_TRANS_HALF_TOPIC 改为真实 Topic
            MessageExtBrokerInner msgInner = endMessageTransaction(result.getPrepareMessage());
            msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
            msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());
            msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());
            msgInner.setStoreTimestamp(result.getPrepareMessage().getStoreTimestamp());
            // 将消息存储到真实 Topic 中，供 Consumer 消费
            RemotingCommand sendResult = sendFinalMessage(msgInner);
            if (sendResult.getCode() == ResponseCode.SUCCESS) {
                // 将消息存储到 RMQ_SYS_TRANS_OP_HALF_TOPIC，标记为删除状态，事务消息回查的定时任务中会做处理
                this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
            }
            return sendResult;
        }
        return res;
    }
} else if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) {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;
    }
}
  1. RocketMQ Broker 端定时任务回查数据库事务部分

方法入口：org.apache.rocketmq.broker.transaction.TransactionalMessageCheckService

@Override
protected void onWaitEnd() {long timeout = brokerController.getBrokerConfig().getTransactionTimeOut();
    // 超过 15 次的回查事务状态失败后，默认是丢弃此消息
    int checkMax = brokerController.getBrokerConfig().getTransactionCheckMax();
    long begin = System.currentTimeMillis();
    log.info("Begin to check prepare message, begin time:{}", begin);
    this.brokerController.getTransactionalMessageService().check(timeout, checkMax, this.brokerController.getTransactionalMessageCheckListener());
    log.info("End to check prepare message, consumed time:{}", System.currentTimeMillis() - begin);
}

回查事务调用入口：

// 此段代码为 TransactionalMessageServiceImpl 类中的 check 方法
List<MessageExt> opMsg = pullResult.getMsgFoundList();
boolean isNeedCheck = (opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime)
    || (opMsg != null && (opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout))
    || (valueOfCurrentMinusBorn <= -1);

if (isNeedCheck) {if (!putBackHalfMsgQueue(msgExt, i)) {continue;}
    // 调用 AbstractTransactionalMessageCheckListener 的
    listener.resolveHalfMsg(msgExt);
} else {pullResult = fillOpRemoveMap(removeMap, opQueue, pullResult.getNextBeginOffset(), halfOffset, doneOpOffset);
    log.info("The miss offset:{} in messageQueue:{} need to get more opMsg, result is:{}", i,
        messageQueue, pullResult);
    continue;
}

// 此方法在 AbstractTransactionalMessageCheckListener 类中
public void resolveHalfMsg(final MessageExt msgExt) {executorService.execute(new Runnable() {
        @Override
        public void run() {
            try {sendCheckMessage(msgExt);
            } catch (Exception e) {LOGGER.error("Send check message error!", e);
            }
        }
    });
}

// 此方法在 AbstractTransactionalMessageCheckListener 类中
public void sendCheckMessage(MessageExt msgExt) throws Exception {CheckTransactionStateRequestHeader checkTransactionStateRequestHeader = new CheckTransactionStateRequestHeader();
    checkTransactionStateRequestHeader.setCommitLogOffset(msgExt.getCommitLogOffset());
    checkTransactionStateRequestHeader.setOffsetMsgId(msgExt.getMsgId());
    checkTransactionStateRequestHeader.setMsgId(msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
    checkTransactionStateRequestHeader.setTransactionId(checkTransactionStateRequestHeader.getMsgId());
    checkTransactionStateRequestHeader.setTranStateTableOffset(msgExt.getQueueOffset());
    msgExt.setTopic(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC));
    msgExt.setQueueId(Integer.parseInt(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_QUEUE_ID)));
    msgExt.setStoreSize(0);
    String groupId = msgExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);
    Channel channel = brokerController.getProducerManager().getAvaliableChannel(groupId);
    if (channel != null) {
        // 通过 Netty 发送请求到 RocketMQ Client 端，执行 checkTransactionState 方法
        brokerController.getBroker2Client().checkProducerTransactionState(groupId, channel, checkTransactionStateRequestHeader, msgExt);
    } else {LOGGER.warn("Check transaction failed, channel is null. groupId={}", groupId);
    }
}

RocketMQ Client 接收到服务端的请求后，重新调用回查数据库事务方法，并将事务结果再次提交到 RocketMQ Broker 端
方法入口：org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl 类的方法

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);
    } else {log.warn("CheckTransactionState, pick transactionListener by group[{}] failed", group);
    }
} catch (Throwable e) {log.error("Broker call checkTransactionState, but checkLocalTransactionState exception", e);
    exception = e;
}

this.processTransactionState(
    localTransactionState,
    group,
    exception);

六、补充一个问题

官网有提及，事务消息是不支持延迟消息和批量消息，我手贱试了一下延迟消息，事务消息设置一个 DelayTimeLevel，结果这条消息就一直无法从 RMQ_SYS_TRANS_HALF_TOPIC 移除掉了，应用模块的日志发现在反复地尝试回查事务，Console 界面上 RMQ_SYS_TRANS_HALF_TOPIC 的消息查询列表很快就超过 2000 条记录了，为什么？

我们回到代码层面进行分析，过程如下：

1. 设置了 DelayTimeLevel 后，数据事务提交后（或是回查数据库事务完成后），将消息写入目标 Topic 时，由于 DelayTimeLevel 的干扰，目标 Topic 将变成 SCHEDULE_TOPIC_XXXX，同时 REAL_TOPIC 变成 RMQ_SYS_TRANS_HALF_TOPIC，真实的 Topic 在这个环节已经丢失。

// RocketMQ Broker 端接受事务提交后的处理
org.apache.rocketmq.broker.processor.EndTransactionProcessor 类
OperationResult result = new OperationResult();
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {
    // 这里调用 CommitLog 的 putMessage 方法
    result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);
    if (result.getResponseCode() == ResponseCode.SUCCESS) {RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
        if (res.getCode() == ResponseCode.SUCCESS) {
            // 修改消息的 Topic 为由 RMQ_SYS_TRANS_HALF_TOPIC 改为真实 Topic
            MessageExtBrokerInner msgInner = endMessageTransaction(result.getPrepareMessage());
            msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
            msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());
            msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());
            msgInner.setStoreTimestamp(result.getPrepareMessage().getStoreTimestamp());
            // 将消息存储到真实 Topic 中，此时 Topic 已经变成 SCHEDULE_TOPIC_XXXX
            RemotingCommand sendResult = sendFinalMessage(msgInner);
            if (sendResult.getCode() == ResponseCode.SUCCESS) {
                // 将消息存储到 RMQ_SYS_TRANS_OP_HALF_TOPIC，标记为删除状态，事务消息回查的定时任务中会做处理
                this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
            }
            return sendResult;
        }
        return res;
    }
}

// 此段代码在 org.apache.rocketmq.store.CommitLog 类的 putMessage 方法中
// 由于 DelayTimeLevel 的干扰，目标 Topic 将变成 SCHEDULE_TOPIC_XXXX
final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
    || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
    // Delay Delivery
    if (msg.getDelayTimeLevel() > 0) {if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
        }

        topic = ScheduleMessageService.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);
    }
}

打印的日志示例如下：

2019-10-17 14\:41\:05 INFO EndTransactionThread_4 - Transaction op message write successfully. messageId=0A00851D00002A9F0000000000000E09, queueId=0 
msgExt:MessageExt [queueId=0, storeSize=335, queueOffset=5, sysFlag=8, bornTimestamp=1571293959305, bornHost=/10.0.133.29:54634, storeTimestamp=1571294460555, 
storeHost=/10.0.133.29:10911, msgId=0A00851D00002A9F0000000000000E09, commitLogOffset=3593, bodyCRC=1849408413, reconsumeTimes=0, preparedTransactionOffset=0, 
toString()=Message{topic='SCHEDULE_TOPIC_XXXX', flag=0, properties={REAL_TOPIC=RMQ_SYS_TRANS_HALF_TOPIC, TRANSACTION_CHECK_TIMES=3, KEYS=msg-test-3, 
TRAN_MSG=true, UNIQ_KEY=0A00851D422C18B4AAC25584B0880000, WAIT=false, DELAY=1, PGROUP=transactionMQProducer, TAGS=transactionTest, REAL_QID=0}, 
body=[72, 101, 108, 108, 111, 84, 105, 109, 101, 58, 51], transactionId='null'}]

2. 延迟消息是定时任务触发的，我刚刚设置的延迟是 1 秒，定时任务又把消息重新放回 RMQ_SYS_TRANS_HALF_TOPIC 中，注意此时只有 RMQ_SYS_TRANS_HALF_TOPIC 有消息，RMQ_SYS_TRANS_OP_HALF_TOPIC 队列是没有这条消息的，如下代码：

// 此段代码在 org.apache.rocketmq.store.schedule.ScheduleMessageService 类 executeOnTimeup 方法内
try {
    // 消息重新回到 RMQ_SYS_TRANS_HALF_TOPIC 队列中
    MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
    PutMessageResult putMessageResult =
        ScheduleMessageService.this.writeMessageStore
            .putMessage(msgInner);

    if (putMessageResult != null
        && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {continue;} else {
        log.error("ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}",
            msgExt.getTopic(), msgExt.getMsgId());
        ScheduleMessageService.this.timer.schedule(
            new DeliverDelayedMessageTimerTask(this.delayLevel,
                nextOffset), DELAY_FOR_A_PERIOD);
        ScheduleMessageService.this.updateOffset(this.delayLevel,
            nextOffset);
        return;
    }
} catch (Exception e) {
    log.error(
        "ScheduleMessageService, messageTimeup execute error, drop it. msgExt="
            + msgExt + ", nextOffset=" + nextOffset + ",offsetPy="
            + offsetPy + ",sizePy=" + sizePy, e);
}

3. 事务消息定时任务启动，查 RMQ_SYS_TRANS_HALF_TOPIC 有消息，但 RMQ_SYS_TRANS_OP_HALF_TOPIC 没有消息，为了保证消息顺序写入，又将此消息重新填入 RMQ_SYS_TRANS_OP_HALF_TOPIC 中，并且触发一次回查事务操作。示例代码如上文回查事务调用入口相同：

// 此段代码为 TransactionalMessageServiceImpl 类中的 check 方法
List<MessageExt> opMsg = pullResult.getMsgFoundList();
boolean isNeedCheck = (opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime)
    || (opMsg != null && (opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout))
    || (valueOfCurrentMinusBorn <= -1);

if (isNeedCheck) {if (!putBackHalfMsgQueue(msgExt, i)) {continue;}
    listener.resolveHalfMsg(msgExt);
} else {pullResult = fillOpRemoveMap(removeMap, opQueue, pullResult.getNextBeginOffset(), halfOffset, doneOpOffset);
    log.info("The miss offset:{} in messageQueue:{} need to get more opMsg, result is:{}", i,
        messageQueue, pullResult);
    continue;
}

这样构成了一个死循环，直到尝试到 15 次才丢弃此消息（默认最大尝试次数是 15 次），这个代价有点大。针对此问题的优化，已经提交 PR 到 RocketMQ 社区，新版本发布后，事务消息将屏蔽 DelayTimeLevel，这个问题就不会再出现了。

在新版本发布之前，我们的解决办法：

  1. 明确研发过程中事务消息禁止设置 DelayTimeLevel。

感觉有风险，毕竟新来的童鞋，不是特别了解此部分功能的可能会手抖加上（像我最早那样）。

  1. 对 RocketMQ Client 做一次简单的封装，比如提供一个 rocketmq-spring-boot-starter，在提供发送事务消息的方法里不提供设置的入口，如下示例：
/**
 * 事务消息发送
 * 不支持延迟发送和批量发送
 */
public void sendMessageInTransaction(String topic, String tag, Object message, String requestId) throws Exception {TransactionMQProducer producer = annotationScan.getProducer(topic + "_" + tag);
   producer.sendMessageInTransaction(MessageBuilder.of(topic, tag, message, requestId).build(), message);
}

应该靠谱一些，毕竟从源头杜绝了 DelayTimeLevel 参数的设置。

七、结束语

本篇简单介绍了事务消息的解决的场景和职责的界限，基本的设计思路和流程，在此借鉴学习了 RocketMQ 作者的图稿，然后挑了部分代码作简要的讲解，还是自己的刨坑过程，文章内有任何不正确或不详尽之处请留言指导，谢谢。

专注 Java 高并发、分布式架构，更多技术干货分享与心得，请关注公众号：Java 架构社区

正文完
apache github
发表至： java
2019-10-19
 0
关于java:java虚拟机运行时数据区域
Java-最常见-200-面试题全解析面试必备附答案
关于java:MySQL数据库基础入门教程7天MySQL从入门到精通视频资料作业题
Android 自定义 View 之 LeavesLoading