沈阳网站制作思路,58同城推广电话,wordpress 登录失败,石家庄网站推广方案什么是rocketmq事务消息
事务消息是 Apache RocketMQ 提供的一种高级消息类型#xff0c;支持在分布式场景下保障消息生产和本地事务的最终一致性。
RocketMQ的分布式事务又称为“半消息事务”。
事务消息处理流程
RocketMQ是靠半消息机制实现分布式事务
事务消息#x…什么是rocketmq事务消息
事务消息是 Apache RocketMQ 提供的一种高级消息类型支持在分布式场景下保障消息生产和本地事务的最终一致性。
RocketMQ的分布式事务又称为“半消息事务”。
事务消息处理流程
RocketMQ是靠半消息机制实现分布式事务
事务消息MQ 提供类似 X/Open XA 的分布事务功能通过 MQ 事务消息能达到分布式事务的最终一致。
半消息暂不能投递的消息发送方已经将消息成功发送到了 MQ 服务端但是服务端未收到生产者对该消息的二次确认此时该消息被标记成“暂不能投递”状态处于该种状态下的消息即半消息。
半消息回查由于网络闪断、生产者应用重启等原因导致某条事务消息的二次确认丢失MQ 服务端通过扫描发现某条消息长期处于“半消息”时需要主动向消息生产者询问该消息的最终状态Commit 或是 Rollback该过程即消息回查。
事务消息交互流程如下图所示。 1. 生产者将消息发送至Apache RocketMQ服务端。
2. Apache RocketMQ服务端将消息持久化成功之后向生产者返回Ack确认消息已经发送成功此时消息被标记为暂不能投递这种状态下的消息即为半事务消息。
3. 生产者开始执行本地事务逻辑。
4. 生产者根据本地事务执行结果向服务端提交二次确认结果Commit或是Rollback服务端收到确认结果后处理逻辑如下
• 二次确认结果为Commit服务端将半事务消息标记为可投递并投递给消费者。
• 二次确认结果为Rollback服务端将回滚事务不会将半事务消息投递给消费者。
5. 在断网或者是生产者应用重启的特殊情况下若服务端未收到发送者提交的二次确认结果或服务端收到的二次确认结果为Unknown未知状态经过固定时间后服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。 说明 服务端回查的间隔时间和最大回查次数请参见参数限制。
6. 生产者收到消息回查后需要检查对应消息的本地事务执行的最终结果。
7. 生产者根据检查到的本地事务的最终状态再次提交二次确认服务端仍按照步骤4对半事务消息进行处理。
事务消息生命周期 事务消息
• 初始化半事务消息被生产者构建并完成初始化待发送到服务端的状态。
• 事务待提交半事务消息被发送到服务端和普通消息不同并不会直接被服务端持久化而是会被单独存储到事务存储系统中等待第二阶段本地事务返回执行结果后再提交。此时消息对下游消费者不可见。
• 消息回滚第二阶段如果事务执行结果明确为回滚服务端会将半事务消息回滚该事务消息流程终止。
• 提交待消费第二阶段如果事务执行结果明确为提交服务端会将半事务消息重新存储到普通存储系统中此时消息对下游消费者可见等待被消费者获取并消费。
• 消费中消息被消费者获取并按照消费者本地的业务逻辑进行处理的过程。此时服务端会等待消费者完成消费并提交消费结果如果一定时间后没有收到消费者的响应Apache RocketMQ会对消息进行重试处理。
• 消费提交消费者完成消费处理并向服务端提交消费结果服务端标记当前消息已经被处理包括消费成功和失败。Apache RocketMQ默认支持保留所有消息此时消息数据并不会立即被删除只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前消费者仍然可以回溯消息重新消费。
• 消息删除Apache RocketMQ按照消息保存机制滚动清理最早的消息数据将消息从物理文件中删除。
示例
下面是使用 RocketMQ 实现事务的一个例子
生产者实现事务监听器
首先需要实现一个 RocketMQ 的事务监听器接口RocketMQLocalTransactionListener这个接口定义了在发送和确认事务消息时的回调方法。您需要根据业务逻辑来实现这些方法。
executeLocalTransaction 方法
这个方法在发送事务消息时被调用用于执行本地事务。具体步骤如下
1. 获取消息中的事务 ID。
2. 根据事务索引来模拟本地事务执行的状态。
3. 将事务状态放入 localTrans 映射中以备后续 checkLocalTransaction 方法使用。
根据您的代码executeLocalTransaction 方法中模拟了三种状态
• 如果状态为 0表示本地事务成功返回 RocketMQLocalTransactionState.COMMIT消息将被提交。
• 如果状态为 1表示本地事务失败返回 RocketMQLocalTransactionState.ROLLBACK消息将被回滚。
• 如果状态为 2表示本地事务状态未知返回 RocketMQLocalTransactionState.UNKNOWN。
checkLocalTransaction 方法
这个方法在消息的确认状态时被调用用于检查本地事务的状态。具体步骤如下 获取消息中的事务 ID。 根据之前保存在 localTrans 映射中的事务状态决定消息的提交、回滚或未知。
checkLocalTransaction 方法会根据之前在 executeLocalTransaction 方法中保存的状态来返回相应的事务状态。
RocketMQTransactionListener
public class TransactionListenerImpl implements RocketMQLocalTransactionListener { private AtomicInteger transactionIndex new AtomicInteger(0); private ConcurrentHashMapString, Integer localTrans new ConcurrentHashMapString, Integer(); Override public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { String transId (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID); System.out.printf(#### executeLocalTransaction is executed, msgTransactionId%s %n, transId); int value transactionIndex.getAndIncrement(); int status value % 3; localTrans.put(transId, status); if (status 0) { // Return local transaction with success(commit), in this case, // this message will not be checked in checkLocalTransaction() System.out.printf( # COMMIT # Simulating msg %s related local transaction exec succeeded! ### %n, msg.getPayload()); return RocketMQLocalTransactionState.COMMIT; } if (status 1) { // Return local transaction with failure(rollback) , in this case, // this message will not be checked in checkLocalTransaction() System.out.printf( # ROLLBACK # Simulating %s related local transaction exec failed! %n, msg.getPayload()); return RocketMQLocalTransactionState.ROLLBACK; } System.out.printf( # UNKNOW # Simulating %s related local transaction exec UNKNOWN! \n); return RocketMQLocalTransactionState.UNKNOWN; } Override public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { String transId (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID); RocketMQLocalTransactionState retState RocketMQLocalTransactionState.COMMIT; Integer status localTrans.get(transId); if (null ! status) { switch (status) { case 0: retState RocketMQLocalTransactionState.COMMIT; break; case 1: retState RocketMQLocalTransactionState.ROLLBACK; break; case 2: retState RocketMQLocalTransactionState.UNKNOWN; break; } } System.out.printf(------ !!! checkLocalTransaction is executed once, msgTransactionId%s, TransactionState%s status%s %n, transId, retState, status); return retState; }
}
消费者
Service
RocketMQMessageListener(topic ${demo.rocketmq.transTopic}, consumerGroup string_trans_consumer)
public class StringTransactionalConsumer implements RocketMQListenerString { Override public void onMessage(String message) { System.out.printf(------- StringTransactionalConsumer received: %s \n, message); }
}
这些步骤基本上涵盖了使用 RocketMQ 实现事务的主要过程。可以根据具体的业务需求和环境进行调整和配置。
总结
使用半消息实现分布式事务在提供分布式事务支持和保证消息传递的原子性方面具有优势但需要引入MQ并提供查询事务接口。在选择是否使用半消息实现分布式事务时需要根据具体的业务需求和系统性能要求来进行权衡和选择。
