一个专门做破解的网站,中国企业500强第一名是谁,wordpress怎么导入自己的php,开发网站公司收费参考#xff1a;https://rocketmq.apache.org/zh/docs/featureBehavior/04transactionmessage/ 文章目录 概要错误的方案方案一#xff1a;业务方自己实现方案二#xff1a;RocketMQ 事务消息什么是事务消息事务消息处理流程事务消息生命周期使用限制使用示例使用建议 概要 … 参考https://rocketmq.apache.org/zh/docs/featureBehavior/04transactionmessage/ 文章目录 概要错误的方案方案一业务方自己实现方案二RocketMQ 事务消息什么是事务消息事务消息处理流程事务消息生命周期使用限制使用示例使用建议 概要
说起分布式事务就会谈到 Bob 给 Smith 账户转账的问题2 个账号 分别处于 A、B 两个 DB或者说 2 个不同的子系统中Bob 的账户要扣钱Smith 的账户要加钱如何保证原子性
一般思路是通过消息中间件来实现 “最终一致性”A系统扣钱然后发送消息给 MQB系统接收此消息进行加钱。
但这里有个问题是先update DB后发消息还是先发消息后update DB呢
假设先update DB成功发送消息时网络问题重发又失败怎么办假设先发送消息成功update DB失败。消息已经发出去了又不能撤回怎么办
所以只要发送消息和update DB这2个操作不是原子的无论谁先谁后都会有问题。
错误的方案
有人可能会想到可以把 “发送消息” 这个网络调用和 update DB 放在同一个事务中如果发送消息失败update DB 自动回滚。这样不就保证原子性了吗
这个方案看似正确但其实是错误的原因
把网络调用放在 DB 事务中可能会因为网络的延时导致 DB 长事务。严重的还会 block 整个 DB这个风险很大。如果发送消息失败发送方并不知道是 MQ 没收到消息还是已收到消息只是返回 response 的时候失败了。如果接收方已经收到消息了而发送方认为没有收到执行 update DB 回滚的操作会导致 A 账户钱没有扣B 账户的钱却增加了。
方案一业务方自己实现
假设消息中间件没提供 “事务消息” 功能比如 Kafka。那该如何解决这个问题
结局方案如下
Producer 准备 1 个消息表把 update DB 和 send message 这 2 个操作放在一个DB事务中。准备一个后台程序不停地把消息表中的 message 传送给消息中间件。发送失败则不断重试直到成功。允许消息重复但消息不会丢。Consumer 准备 1 个判重表。处理过的消息记录在里面实现业务的幂等性
通过以上 3 步就基本解决了原子性问题
但此方案缺点是需要设计 DB 消息表同时还需要 1 个后台任务不断扫描本地消息。导致消息的处理和业务逻辑耦合额外增加业务方的负担。
方案二RocketMQ 事务消息
目前各大知名电商平台和互联网公司几乎都是使用 “事务消息” 这种方案来实现 “最终一致性” 的。这种方式适用的业务场景广泛且比较靠谱。
什么是事务消息
事务消息是 Apache RocketMQ 提供的一种高级消息类型支持在分布式场景下保障消息生产和本地事务的最终一致性。
事务消息处理流程
事务消息交互流程如下图所示。 生产者将消息发送至Apache RocketMQ服务端。 Apache RocketMQ服务端将消息持久化成功之后向生产者返回Ack确认消息已经发送成功此时消息被标记为暂不能投递这种状态下的消息即为半事务消息。 生产者开始执行本地事务逻辑。 生产者根据本地事务执行结果向服务端提交二次确认结果Commit或是Rollback服务端收到确认结果后处理逻辑如下 二次确认结果为Commit服务端将半事务消息标记为可投递并投递给消费者。 二次确认结果为Rollback服务端将回滚事务不会将半事务消息投递给消费者。 在断网或者是生产者应用重启的特殊情况下若服务端未收到发送者提交的二次确认结果或服务端收到的二次确认结果为Unknown未知状态经过固定时间后服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查服务端回查的间隔时间和最大回查次数请参见参数限制。 生产者收到消息回查后需要检查对应消息的本地事务执行的最终结果。 生产者根据检查到的本地事务的最终状态再次提交二次确认服务端仍按照步骤4对半事务消息进行处理。
事务消息生命周期 初始化半事务消息被生产者构建并完成初始化待发送到服务端的状态。 事务待提交半事务消息被发送到服务端和普通消息不同并不会直接被服务端持久化而是会被单独存储到事务存储系统中等待第二阶段本地事务返回执行结果后再提交。此时消息对下游消费者不可见。 消息回滚第二阶段如果事务执行结果明确为回滚服务端会将半事务消息回滚该事务消息流程终止。 提交待消费第二阶段如果事务执行结果明确为提交服务端会将半事务消息重新存储到普通存储系统中此时消息对下游消费者可见等待被消费者获取并消费。 消费中消息被消费者获取并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果如果一定时间后没有收到消费者的响应Apache RocketMQ会对消息进行重试处理。具体信息请参见消费重试。 消费提交消费者完成消费处理并向服务端提交消费结果服务端标记当前消息已经被处理包括消费成功和失败。 Apache RocketMQ默认支持保留所有消息此时消息数据并不会立即被删除只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前消费者仍然可以回溯消息重新消费。 消息删除Apache RocketMQ按照消息保存机制滚动清理最早的消息数据将消息从物理文件中删除。更多信息请参见消息存储和清理机制。
使用限制 消息类型一致性 事务消息仅支持在 MessageType 为 Transaction 的主题内使用即事务消息只能发送至类型为事务消息的主题中发送的消息的类型必须和主题的类型一致。 消费事务性 Apache RocketMQ 事务消息保证本地主分支事务和下游消息发送事务的一致性但不保证消息消费结果和上游事务的一致性。因此需要下游业务分支自行保证消息正确处理建议消费端做好消费重试如果有短暂失败可以利用重试机制保证最终处理成功。 中间状态可见性 Apache RocketMQ 事务消息为最终一致性即在消息提交到下游消费端处理完成之前下游分支和上游事务之间的状态会不一致。因此事务消息仅适合接受异步执行的事务场景。 事务超时机制 Apache RocketMQ 事务消息的生命周期存在超时机制即半事务消息被生产者发送服务端后如果在指定时间内服务端无法确认提交或者回滚状态则消息默认会被回滚。事务超时时间请参见参数限制。
使用示例
创建主题
Apache RocketMQ 5.0版本下创建主题操作推荐使用mqadmin工具需要注意的是对于消息类型需要通过属性参数添加。示例如下
sh mqadmin updateTopic -n nameserver_address -t topic_name -c cluster_name -a message.typeTRANSACTION
发送消息
事务消息相比普通消息发送时需要修改以下几点 发送事务消息前需要开启事务并关联本地的事务执行。 为保证事务一致性在构建生产者时必须设置事务检查器和预绑定事务消息发送的主题列表客户端内置的事务检查器会对绑定的事务主题做异常状态恢复。
创建事务主题
NORMAL类型Topic不支持TRANSACTION类型消息生产消息会报错。
./bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster -a message.typeTRANSACTION
-c 集群名称-t Topic名称-n nameserver地址-a 额外属性本例给主题添加了message.type为TRANSACTION的属性用来支持事务消息
以Java语言为例使用事务消息示例参考如下 //演示demo模拟订单表查询服务用来确认订单事务是否提交成功。private static boolean checkOrderById(String orderId) {return true;}//演示demo模拟本地事务的执行结果。private static boolean doLocalTransaction() {return true;}public static void main(String[] args) throws ClientException {ClientServiceProvider provider new ClientServiceProvider();MessageBuilder messageBuilder new MessageBuilderImpl();//构造事务生产者事务消息需要生产者构建一个事务检查器用于检查确认异常半事务的中间状态。Producer producer provider.newProducerBuilder().setTransactionChecker(messageView - {/*** 事务检查器一般是根据业务的ID去检查本地事务是否正确提交还是回滚此处以订单ID属性为例。* 在订单表找到了这个订单说明本地事务插入订单的操作已经正确提交如果订单表没有订单说明本地事务已经回滚。*/final String orderId messageView.getProperties().get(OrderId);if (Strings.isNullOrEmpty(orderId)) {// 错误的消息直接返回Rollback。return TransactionResolution.ROLLBACK;}return checkOrderById(orderId) ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;}).build();//开启事务分支。final Transaction transaction;try {transaction producer.beginTransaction();} catch (ClientException e) {e.printStackTrace();//事务分支开启失败直接退出。return;}Message message messageBuilder.setTopic(topic)//设置消息索引键可根据关键字精确查找某条消息。.setKeys(messageKey)//设置消息Tag用于消费端根据指定Tag过滤消息。.setTag(messageTag)//一般事务消息都会设置一个本地事务关联的唯一ID用来做本地事务回查的校验。.addProperty(OrderId, xxx)//消息体。.setBody(messageBody.getBytes()).build();//发送半事务消息final SendReceipt sendReceipt;try {sendReceipt producer.send(message, transaction);} catch (ClientException e) {//半事务消息发送失败事务可以直接退出并回滚。return;}/*** 执行本地事务并确定本地事务结果。* 1. 如果本地事务提交成功则提交消息事务。* 2. 如果本地事务提交失败则回滚消息事务。* 3. 如果本地事务未知异常则不处理等待事务消息回查。*/boolean localTransactionOk doLocalTransaction();if (localTransactionOk) {try {transaction.commit();} catch (ClientException e) {// 业务可以自身对实时性的要求选择是否重试如果放弃重试可以依赖事务消息回查机制进行事务状态的提交。e.printStackTrace();}} else {try {transaction.rollback();} catch (ClientException e) {// 建议记录异常信息回滚异常时可以无需重试依赖事务消息回查机制进行事务状态的提交。e.printStackTrace();}}}使用建议 避免大量未决事务导致超时 Apache RocketMQ支持在事务提交阶段异常的情况下发起事务回查保证事务一致性。但生产者应该尽量避免本地事务返回未知结果。大量的事务检查会导致系统性能受损容易导致事务处理延迟。 正确处理进行中的事务 消息回查时对于正在进行中的事务不要返回Rollback或Commit结果应继续保持Unknown的状态。 一般出现消息回查时事务正在处理的原因为事务执行较慢消息回查太快。解决方案如下 将第一次事务回查时间设置较大一些但可能导致依赖回查的事务提交延迟较大。 程序能正确识别正在进行中的事务。
