网站建设和网页设计pdf,wordpress 免费cdn,电子商务网站建设与管理考试,怎么开通网站平台RocketMQ提供了事务消息的功能#xff0c;采用2PC(两阶段协议)补偿机制#xff08;事务回查#xff09;的分布式事务功能#xff0c;通过这种方式能达到分布式事务的最终一致。
一. 概述
半事务消息#xff1a;指的是发送至broker但是还没被commit的消息#xff0c;在半…RocketMQ提供了事务消息的功能采用2PC(两阶段协议)补偿机制事务回查的分布式事务功能通过这种方式能达到分布式事务的最终一致。
一. 概述
半事务消息指的是发送至broker但是还没被commit的消息在半事务消息被确认之前都是无法被消费者消费的。消息回查由于网络闪断、生产者应用重启等原因导致某条事务消息的二次确认丢失broker 通过扫描发现某条消息长期处于“半事务消息”时需要主动向消息生产者询问该消息的最终状态commit 或是 rollback该询问过程即消息回查。
二. 交互流程 事务消息发送步骤如下 发送方将半事务消息发送至broker。 broker将消息持久化成功之后向发送方返回 Ack确认消息已经发送成功此时消息为半事务消息。 发送方开始执行本地事务逻辑。 发送方根据本地事务执行结果向服务端提交二次确认commit 或是 rollback服务端收到 commit 状态则将半事务消息标记为可投递订阅方最终将收到该消息服务端收到 rollback 状态则“删除”半事务消息订阅方将不会接受该消息。 在断网或者是应用重启的特殊情况下上述步骤 4 提交的二次确认最终未到达服务端经过固定时间后服务端将对该消息发起消息回查。 发送方收到消息回查后需要检查对应消息的本地事务执行的最终结果。 发送方根据检查得到的本地事务的最终状态再次提交二次确认服务端仍按照步骤 4 对半事务消息进行操作。
三. 示例代码
public static void main(String[] args) throws MQClientException, InterruptedException {TransactionMQProducer producer new TransactionMQProducer(please_rename_unique_group_name);ExecutorService executorService new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueueRunnable(2000), new ThreadFactory() {Overridepublic Thread newThread(Runnable r) {Thread thread new Thread(r);thread.setName(client-transaction-msg-check-thread);return thread;}});producer.setExecutorService(executorService);producer.setTransactionListener(new TransactionListener() {Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {// 执行业务代码 ....// 最终返回业务代码的事务状态return LocalTransactionState.COMMIT_MESSAGE;}Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {// 回查本地事务状态return LocalTransactionState.ROLLBACK_MESSAGE;}});producer.start();String[] tags new String[] {TagA, TagB, TagC, TagD, TagE};for (int i 0; i 10; i) {try {Message msg new Message(TopicTest1234, tags[i % tags.length], KEY i,(Hello RocketMQ i).getBytes(RemotingHelper.DEFAULT_CHARSET));SendResult sendResult producer.sendMessageInTransaction(msg, null);System.out.printf(%s%n, sendResult);Thread.sleep(10);} catch (MQClientException | UnsupportedEncodingException e) {e.printStackTrace();}}for (int i 0; i 100000; i) {Thread.sleep(1000);}producer.shutdown();
}TransactionMQProducer 支持事务消息的生产者继承自 DefaultMQProducer在默认生产者上进行了扩展支持发送事务消息。它拥有一个线程池 executorService 用来异步执行本地事务和回查事务还需要注册 TransactionListener 事务监听器里面包含了执行本地事务和回查事务的逻辑。
三. 源码分析
整个事务消息的处理流程可以分为以下五个步骤
Half消息发送Half消息存储提交事务状态处理事务状态事务回查
3.1 Half消息发送
除事务回查外事务消息的时序图大致如下 TransactionMQProducer 是RocketMQ提供的支持发送事务消息的生产者它继承自 DefaultMQProducer也是一个外观类代码非常的简单核心逻辑依然在 DefaultMQProducerImpl属性如下
public class TransactionMQProducer extends DefaultMQProducer {// 事务回查监听private TransactionCheckListener transactionCheckListener;// 回查线程池最小线程数private int checkThreadPoolMinSize 1;// 回查线程池最大线程数private int checkThreadPoolMaxSize 1;// 最大回查请求数阻塞队列容量private int checkRequestHoldMax 2000;// 执行本地事务/事务回查的线程池private ExecutorService executorService;// 事务监听器:本地事务、事务回查逻辑private TransactionListener transactionListener;
}启动 TransactionMQProducer必须先注册 TransactionListener实现本地事务的执行逻辑和事务回查逻辑Producer在发送Half消息成功后会自动执行executeLocalTransaction在Broker请求事务回查时自动执行checkLocalTransaction。
然后Producer就可以启动了在启动默认Producer之前会对checkExecutor和checkRequestQueue进行初始化如果没有设置线程池会自动创建。
public void initTransactionEnv() {TransactionMQProducer producer (TransactionMQProducer) this.defaultMQProducer;if (producer.getExecutorService() ! null) {this.checkExecutor producer.getExecutorService();} else {// 事务回查请求队列this.checkRequestQueue new LinkedBlockingQueueRunnable(producer.getCheckRequestHoldMax());// 事务回查线程池this.checkExecutor new ThreadPoolExecutor(producer.getCheckThreadPoolMinSize(),producer.getCheckThreadPoolMaxSize(),1000 * 60,TimeUnit.MILLISECONDS,this.checkRequestQueue);}
}初始化完成以后就是Producer的正常启动逻辑这里不再赘述。
发送事务消息对应的方法是sendMessageInTransaction
// org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendMessageInTransaction
public TransactionSendResult sendMessageInTransaction(final Message msg,final LocalTransactionExecuter localTransactionExecuter, final Object arg)throws MQClientException {// 判断检查本地事务Listener是否存在TransactionListener transactionListener getCheckListener();if (null localTransactionExecuter null transactionListener) {throw new MQClientException(tranExecutor is null, null);}// ignore DelayTimeLevel parameterif (msg.getDelayTimeLevel() ! 0) {MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);}// 检查消息内容Validators.checkMessage(msg, this.defaultMQProducer);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);}LocalTransactionState localTransactionState LocalTransactionState.UNKNOW;Throwable localException null;switch (sendResult.getSendStatus()) {case SEND_OK: {try {if (sendResult.getTransactionId() ! null) {msg.putUserProperty(__transactionId__, sendResult.getTransactionId());}String transactionId msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);if (null ! transactionId !.equals(transactionId)) {msg.setTransactionId(transactionId);}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;}if (localTransactionState ! LocalTransactionState.COMMIT_MESSAGE) {log.info(executeLocalTransactionBranch return {}, localTransactionState);log.info(msg.toString());}} 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;}try {// 执行endTransaction方法如果半消息发送失败或本地事务执行失败告诉服务端是删除半消息// 半消息发送成功且本地事务执行成功则告诉broker提交半消息this.endTransaction(sendResult, localTransactionState, localException);} catch (Exception e) {log.warn(local transaction execute localTransactionState , but end broker transaction failed, e);}// .......
}它主要做了以下事情
设置属性TRAN_MSGtrue同步发送Half消息消息发送成功执行本地事务提交本地事务状态
Tips事务消息不支持延时会在发送前自动忽略延迟级别。
if (msg.getDelayTimeLevel() ! 0) {MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
}Broker判断是否是事务消息的依据是通过Properties的TRAN_MSG属性判断的Producer在发送消息前会进行设置。
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, true);消息属性设置完毕调用send方法同步发送给Broker并获取发送状态。
SendResult sendResult this.send(msg); 3.2 Half消息存储
Half消息发送到BrokerBroker要负责存储且此时消息对Consumer是不可见的看看它是如何处理的。
SendMessageProcessor 会对接收到的消息进行判断如果是事务消息会转交给TransactionalMessageService 处理普通消息直接转交给 MessageStore 处理。
// org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage// 是否是事务消息通过TRAN_MSG属性判断
String transFlag origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (transFlag ! null Boolean.parseBoolean(transFlag)) {// 事务消息的处理if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {response.setCode(ResponseCode.NO_PERMISSION);response.setRemark(the broker[ this.brokerController.getBrokerConfig().getBrokerIP1() ] sending transaction message is forbidden);return CompletableFuture.completedFuture(response);}putMessageResult this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
}TransactionalMessageService 使用了桥接模式大部分操作会交给桥接类 TransactionalMessageBridge 执行。在处理Half消息时为了不让Consumer可见会像处理延迟消息一样改写Topic和queueId将消息统一扔到RMQ_SYS_TRANS_HALF_TOPIC这个Topic下默认的queueId为0。同时为了后续消息Commit时重新写入正常消息必须将真实的Topic和queueId等属性先保留到Properties中。
// org.apache.rocketmq.broker.transaction.queue.TransactionalMessageBridge#parseHalfMessageInner
private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {// 真实的Topic和queueId存储到PropertiesMessageAccessor.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_TOPICmsgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());msgInner.setQueueId(0);msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));return msgInner;
}消息的Topic被改写后正常写入CommitLog但不会对Consumer可见。
3.3 提交事务状态
Broker将消息写入CommitLog后会返回结果SendResult如果发送成功Producer开始执行本地事务
// org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendMessageInTransaction
if (sendResult.getTransactionId() ! null) {// 设置事务IDmsg.putUserProperty(__transactionId__, sendResult.getTransactionId());
}
String transactionId msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (null ! transactionId !.equals(transactionId)) {msg.setTransactionId(transactionId);
}
if (null ! localTransactionExecuter) {// 老的本地事务处理已经被废弃localTransactionState localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
} else if (transactionListener ! null) {// 执行本地事务获取事务状态localTransactionState transactionListener.executeLocalTransaction(msg, arg);
}
if (null localTransactionState) {localTransactionState LocalTransactionState.UNKNOW;
}if (localTransactionState ! LocalTransactionState.COMMIT_MESSAGE) {log.info(executeLocalTransactionBranch return {}, localTransactionState);log.info(msg.toString());
}将本地事务执行状态 LocalTransactionState 提交到 Broker方法是endTransaction。先根据 MessageQueue 找到Broker的主机地址然后构建提交事务请求头 EndTransactionRequestHeader 并设置相关属性请求头属性如下
public class EndTransactionRequestHeader implements CommandCustomHeader {// 生产者组private String producerGroup;// ConsumeQueue Offsetprivate Long tranStateTableOffset;// 消息所在CommitLog偏移量private Long commitLogOffset;// 事务状态private Integer commitOrRollback;// 是否Broker发起的回查private Boolean fromTransactionCheck false;// 消息IDprivate String msgId;// 事务IDprivate String transactionId;
}事务状态 commitOrRollback 用数字表示8代表Commit、12代表Rollback、0代表未知状态。请求头构建好以后通过Netty发送数据包给Broker对应的RequestCode为END_TRANSACTION。
public void endTransaction(final Message msg,final SendResult sendResult,final LocalTransactionState localTransactionState,final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {final MessageId id;// 解析MessageId内含消息Offsetif (sendResult.getOffsetMsgId() ! null) {id MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());} else {id MessageDecoder.decodeMessageId(sendResult.getMsgId());}String transactionId sendResult.getTransactionId();// 获取MessageQueue所在Broker的Master主机地址final String brokerAddr this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());// 创建请求头EndTransactionRequestHeader requestHeader new EndTransactionRequestHeader();// 设置事务ID和偏移量requestHeader.setTransactionId(transactionId);requestHeader.setCommitLogOffset(id.getOffset());// 设置事务状态switch (localTransactionState) {case COMMIT_MESSAGE:requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);break;case ROLLBACK_MESSAGE:requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);break;case UNKNOW:requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);break;default:break;}// 执行钩子函数doExecuteEndTransactionHook(msg, sendResult.getMsgId(), brokerAddr, localTransactionState, false);requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());requestHeader.setMsgId(sendResult.getMsgId());String remark localException ! null ? (executeLocalTransactionBranch exception: localException.toString()) : null;// 发送请求this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark,this.defaultMQProducer.getSendMsgTimeout());
}3.4 处理事务状态
Broker通过 EndTransactionProcessor 类来处理 Producer 提交的事务请求首先做校验确保是Master处理该请求因为Slave是没有写权限的
// org.apache.rocketmq.broker.processor.EndTransactionProcessor#processRequest
if (BrokerRole.SLAVE brokerController.getMessageStoreConfig().getBrokerRole()) {response.setCode(ResponseCode.SLAVE_NOT_AVAILABLE);LOGGER.warn(Message store is slave mode, so end transaction is forbidden. );return response;
}然后解析请求头获取事务状态如果是commit操作则根据请求头里的 CommitLogOffset 读取出完整的消息从 Properties 中恢复消息真实的 Topic、queueId 等属性再调用 sendFinalMessage 方法将消息重新写入 CommitLog稍后构建好 ConsumeQueue 消息对 Consumer 就可见了最后调用deletePrepareMessage方法删除Half消息。
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) {// 创建新的Message恢复真实的Topic、queueId等属性重新写入CommitLogMessageExtBrokerInner 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());MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED);RemotingCommand sendResult sendFinalMessage(msgInner);if (sendResult.getCode() ResponseCode.SUCCESS) {/*** 事务消息提交删除Half消息* Half消息不会真的被删除通过写入Op消息来标记它被处理。*/this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());}return sendResult;}return res;}
}如果是Rollback处理就更加简单了因为消息本来就对Consumer是不可见的只需要删除Half消息即可。
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) {// 写入Op消息代表Half消息被处理this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());}return res;}
}实际上Half消息并不会删除因为CommitLog是顺序写的不可能删除单个消息。「删除Half消息」仅仅是给该消息打上一个标记代表它的最终状态已知不需要再回查了。RocketMQ通过引入Op消息来给Half消息打标记Half消息状态确认后会写入一条消息到Op队列对应的Topic为MQ_SYS_TRANS_OP_HALF_TOPIC反之Op队列中不存在的就是状态未确认需要回查的Half消息。
3.5 事务回查
Broker端发起消息事务回查的时序图如下 Half消息写入成功可能因为种种原因没有收到Producer的事务状态提交请求。此时Broker会主动发起事务回查请求给Producer以决定最终将消息Commit还是Rollback。
Half消息最终状态有没有被确认是通过Op队列里的消息判断的。Broker服务启动时会开启TransactionalMessageCheckService线程每隔60秒进行一次消息回查。为了避免消息被无限次的回查RocketMQ通过transactionCheckMax属性设置消息回查的最大次数默认是15次。
// org.apache.rocketmq.broker.transaction.TransactionalMessageCheckService#onWaitEnd
protected void onWaitEnd() {// 回查超时long timeout brokerController.getBrokerConfig().getTransactionTimeOut();// 回查最大次数int checkMax brokerController.getBrokerConfig().getTransactionCheckMax();long begin System.currentTimeMillis();// 开始回查this.brokerController.getTransactionalMessageService().check(timeout, checkMax, this.brokerController.getTransactionalMessageCheckListener());
}回查Half消息时首先要获取Half主题下的所有消息队列。
// 获取RMQ_SYS_TRANS_HALF_TOPIC下所有队列
String topic TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;
SetMessageQueue msgQueues transactionalMessageBridge.fetchMessageQueues(topic);
if (msgQueues null || msgQueues.size() 0) {log.warn(The queue of topic is empty : topic);return;
}然后遍历所有的MessageQueue按个处理所有队列里的待回查的消息。怎么判断消息需要回查呢前面说过了通过Op队列判断因此还需要定位到HalfQueue对应的OpQueue以及它们的ConsumeQueue偏移量。
// 获取对应的 Op队列
MessageQueue opQueue getOpQueue(messageQueue);
// 获取ConsumeQueue Offset
long halfOffset transactionalMessageBridge.fetchConsumeOffset(messageQueue);
long opOffset transactionalMessageBridge.fetchConsumeOffset(opQueue);然后从CommitLog读取出完整的消息。
// 根据offset去CommitLog读取出消息
private GetResult getHalfMsg(MessageQueue messageQueue, long offset) {GetResult getResult new GetResult();PullResult result pullHalfMsg(messageQueue, offset, PULL_MSG_RETRY_NUMBER);getResult.setPullResult(result);ListMessageExt messageExts result.getMsgFoundList();if (messageExts null) {return getResult;}getResult.setMsg(messageExts.get(0));return getResult;
}判断回查次数是否已达上限如果是的话就统一扔到TRANS_CHECK_MAX_TIME_TOPIC下。
if (needDiscard(msgExt, transactionCheckMax) || needSkip(msgExt)) {// 回查次数超过15丢弃消息扔到TRANS_CHECK_MAX_TIME_TOPIClistener.resolveDiscardMsg(msgExt);newOffset i 1;i;continue;
}如果判断消息确实需要回查会调用AbstractTransactionalMessageCheckListener的sendCheckMessage方法恢复消息真实的Topic、queueId等属性然后发回给Producer进行事务的回查确认。
// 发送事务回查消息给Producer
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);// 获取消息生产的GroupIdString groupId msgExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);// 轮询出一台Producer实例Channel channel brokerController.getProducerManager().getAvailableChannel(groupId);if (channel ! null) {// 发送回查请求brokerController.getBroker2Client().checkProducerTransactionState(groupId, channel, checkTransactionStateRequestHeader, msgExt);} else {LOGGER.warn(Check transaction failed, channel is null. groupId{}, groupId);}
}Broker将回查请求发送给Producer后Producer会执行checkLocalTransaction方法检查本地事务然后将事务状态再发送给Broker重复上述流程。
四. 总结
RocketMQ实现事务消息的原理和实现延迟消息的原理类似都是通过改写Topic和queueId暂时将消息先写入一个对Consumer不可见的队列中然后等待Producer执行本地事务提交事务状态后再决定将Half消息Commit或者Rollback。同时可能因为服务宕机或网络抖动等原因Broker没有收到Producer的事务状态提交请求为了对二阶段进行补偿Broker会主动对未确认的Half消息进行事务回查判断消息的最终状态是否确认是通过Op队列实现的Half消息一旦确认事务状态就会往Op队列中写入一条消息消息内容是Half消息所在ConsumeQueue的偏移量。 本文参考转载至 【RocketMQ】事务消息实现原理分析 - 掘金 (juejin.cn) RocketMq之事务消息实现原理 - 掘金 (juejin.cn)
