IDEA2023.1.3破解,IDEA破解,IDEA 2023.1破解,最新IDEA激活码

RocketMQ 实现分布式事务,RocketMQ数据最终一致性

IDEA2023.1.3破解,IDEA破解,IDEA 2023.1破解,最新IDEA激活码

前言

在分布式环境下,经常会有跨服务的事务需求,典型的例子如: 服务A 为账户服务,服务B为包月服务,在服务A扣钱成功之后,要在服务B上做增加包月时长的操作,需要保证数据的一致性。本文主要描写如何在分布式环境下通过Rocket MQ的事务消息保证数据的最终一致性。

注: 最终一致性是指在中间过程中,数据有可能不一致,但经过一段时间后(具体时长视网络延迟以及系统负载而定),数据最终会保持一致。

Rocket MQ事务消息原理

Rocket MQ是阿里开发的一个分布式的开源消息队列组件,目前由Apache开源组织维护,最新版本是4.3.0,已经支持事务消息。

事务消息可以确保本地事务 与 发送消息 之间的原子性,相关概念:

1、Half(Prepare) Message

Producer已经把消息发送给Mq 服务器,但是Mq服务器尚未收到生产者的第二次Ack,这个时候消息会被标记为”temporarily undeliverable”,目前消息的状态为 Half Message

2、Message Status Check

网络断开或者Producer应用重启会导致Mq服务器无法从Producer获取第二次ACK,当Mq服务器发现一个消息长时间处于 Half Message 状态时(默认为60S,可配置),它会主动请求Producer,查询消息Id对应的最新状态(commit 或者 rollback)。

具体流程如下

1、Producer 向Mq服务器 发送消息。

2、Mq服务器收到消息并持久化成功之后,会向 Producer确认首次ACK,此时消息处于 Half Message 状态,并未发送给对应的Consumer。

3、Producer 开始执行本地事务逻辑。

4、根据事务执行结果,Producer 向Mq服务器提交二次确认(commit 或者 rollback)。Mq Server 收到 Commit 状态则将半消息标记为可投递,Consumer 最终将收到该消息;MQ Server 收到 Rollback 状态则删除半消息,Consumer 将不会接受该消息。

5、在断网或者应用重启的情况下,二次ACK未成功的发给Mq Server,Mq Server会主动向 Producer 启动消息回查(Message Status Check),

6、Producer 根据事务执行结果,对消息回查返回对应的结果。

7、Mq Server根据返回结果,决定继续投递消息或者丢弃消息(重复第4步操作)。

流程图如下

事务消息流程.png

实例

针对上面案例,可以通过如下的方式使用事务消息。

Producer:

1、实现 TransactionListener 接口,在executeLocalTransaction方法里执行本地事务逻辑,在checkLocalTransaction方法里返回消息id对应的事务状态,用于Mq的消息回查。

2、通过 TransactionMQProducer构造事务消息并发送。

Consumer:

1、从Mq server获取到消息之后,即开始处理本地事务,处理成功后返回 CONSUME_SUCCESS

2、处理失败则返回 RECONSUME_LATER,Mq server会在稍后重新投递这个消息,又进入步骤1。

注: Consumer 需要做好幂等控制,消息可能会被多次投递到Consumer。

其他

可以看到,极端情况下,可能仍然会出现 消费者出错的情况,不过这种情况建议人工介入处理。对于这种概率非常小的情况,使用人工介入手动处理的办法,比实现一套完整的自动回滚事务系统成本要低很多,要更划算一些。

1、使用限制

以下是使用Rocket mq 事务消息的一些限制:

1、事务消息不支持 delay 或者 batch操作。

2、为了避免一个 Half Message的消息被检查多次 或者 消息积压,默认对每个消息最多进行15次消息回查,可以通过修改broker的 transactionCheckMax 参数来指定次数。如果一个Half Message状态的消息检查次数超过了transactionCheckMax,默认情况下会直接丢弃掉并且打印错误日志,可以通过覆盖 AbstractTransactionCheckListener 类来修改这个行为。

3、通过 transactionMsgTimeout参数可以指定消息回查(Message Status Check)间隔。

4、事务消息可能会被check或者consume多次,要在Consumer端做好幂等控制。

2、事务状态

Rocket MQ里有三种事务状态

1、LocalTransactionState.UNKNOW : 中间状态,意味着Mq server需要稍候再次确认。

2、LocalTransactionState.COMMIT_MESSAGE: 事务完成,意味着消息可以投递给对应的 Consumer。

3、LocalTransactionState.ROLLBACK_MESSAGE: 事务失败,Mq Server会丢弃对应的事务消息,不会投递给对应的Consumer。

代码

1、 Rocket MQ事务消息的设计
http://rocketmq.apache.org/rocketmq/the-design-of-transactional-message/

2、RocketMQ 官方事务消息的例子

https://rocketmq.apache.org/docs/transaction-example/


Warning: A non-numeric value encountered in /data/wangzhan/tech.souyunku.com.wp/wp-content/themes/dux/functions-theme.php on line 1154
赞(95) 打赏



未经允许不得转载:搜云库技术团队 » RocketMQ 实现分布式事务,RocketMQ数据最终一致性

IDEA2023.1.3破解,IDEA破解,IDEA 2023.1破解,最新IDEA激活码
IDEA2023.1.3破解,IDEA破解,IDEA 2023.1破解,最新IDEA激活码

评论 抢沙发

大前端WP主题 更专业 更方便

联系我们联系我们

觉得文章有用就打赏一下文章作者

微信扫一扫打赏

微信扫一扫打赏


Fatal error: Uncaught Exception: Cache directory not writable. Comet Cache needs this directory please: `/data/wangzhan/tech.souyunku.com.wp/wp-content/cache/comet-cache/cache/https/tech-souyunku-com/index.q`. Set permissions to `755` or higher; `777` might be needed in some cases. in /data/wangzhan/tech.souyunku.com.wp/wp-content/plugins/comet-cache/src/includes/traits/Ac/ObUtils.php:367 Stack trace: #0 [internal function]: WebSharks\CometCache\Classes\AdvancedCache->outputBufferCallbackHandler() #1 /data/wangzhan/tech.souyunku.com.wp/wp-includes/functions.php(5109): ob_end_flush() #2 /data/wangzhan/tech.souyunku.com.wp/wp-includes/class-wp-hook.php(303): wp_ob_end_flush_all() #3 /data/wangzhan/tech.souyunku.com.wp/wp-includes/class-wp-hook.php(327): WP_Hook->apply_filters() #4 /data/wangzhan/tech.souyunku.com.wp/wp-includes/plugin.php(470): WP_Hook->do_action() #5 /data/wangzhan/tech.souyunku.com.wp/wp-includes/load.php(1097): do_action() #6 [internal function]: shutdown_action_hook() #7 {main} thrown in /data/wangzhan/tech.souyunku.com.wp/wp-content/plugins/comet-cache/src/includes/traits/Ac/ObUtils.php on line 367