关于分布式事务的随笔[待续]

Posted 2020-09-17 myJavaEE

传统开发中，系统往往是以单体应用形式存在的，没有横跨多个数据库。我们利用关系型数据库自带的事务管理机制就能满足业务中对事务的需求。而大型互联网平台往往是由一系列分布式系统构成的，在SOA和微服务架构盛行的今天，一个看起来简单的功能，内部可能需要调用多个服务(并操作其下的多个数据库)，情况会复杂很多。

在分布式系统中，不可能同时满足“一致性”、“可用性”和“分区容错性”。在互联网的大多数的场景，往往选择牺牲强一致性来换取高可用性，系统只需要保证“最终一致性”，只要这个最终一致的耗时在用户可接受的范围内。

 

强一致性的分布式事务--两阶段提交协议

这里的例子说的是将钱从支付宝转到余额宝中

两阶段提交协议(Two-phase Commit，2PC)由一个事务协调器TC和多个事务执行者Si(操作相应数据库)实现

1) 应用程序（client）发起一个开始请求到TC；
2) TC先将<prepare>消息写到本地日志，之后向所有Si发起<prepare>消息。TC给A的prepare消息是支付宝-1W，TC给B的prepare消息是余额宝+1w。写本地日志是为了故障后恢复
3) Si收到<prepare>消息后，执行具体事务，但不commit，如果成功返回<ready T>，不成功返回<abort T>, 返回前将消息写到日志。
4) TC收集所有执行器返回结果，如果所有执行器都返回<ready T>，则给所有执行器发生送<commit T>消息，执行器收到后执行commit; 如有任一执行器返回<abort T>，那么给所有执行器发送<abort T>消息，执行器收到后执行回滚。同样的发送前将消息写到日志。

需注意
1)发送前将消息写到日志, 以实现故障后恢复
2)X/Open DTP模型是X/Open组织提出的分布式事务模型, TM（事务管理器）和RM（资源管理器）通过XA接口通信. 其通过两阶段提交协议来实现分布式事务. Java平台的事务规范JTA（Java Transaction API）是符合X/Open DTP模型的, 主流关系型数据库也都实现了XA接口. JTA的实现可以使用JBOSS等J2EE容器, 在Tomcat下也可使用Atomikos等独立的JTA框架.
3) 该方案是强一致的, 适合传统应用，比如在同一个方法中跨库操作。但其性能较差(节点间多次通信, 锁资源的时间较长)，并不适合高并发场景。

 

弱一致性的分布式"事务"

已经不是传统意义上的事务了, 原则是牺牲强一致性, 确保最终一致性

1.使用消息队列

1) 支付宝完成扣款，同时记录消息数据(消息记录表为message)，消息数据与业务数据保存在同一数据库中

start transaction
     update A set amount=amount-10000 where userId=1;
     insert into message (userId, amount,status) values(1, 10000, 1);
commit;

commit后，通过实时消息服务将此消息通知余额宝，余额宝处理成功后发送回复成功消息，支付宝收到回复后删除该条消息数据。

 

2)解耦第一种方式

1）支付宝在扣款事务提交之前，向消息服务请求发送消息，消息服务只记录消息数据，而不真正发送(消息事务)；
2）当支付宝扣款事务被提交成功后，向消息服务确认发送; 当支付宝扣款事务提交失败回滚后，向消息服务取消发送. 只有在得到确认发送指令后，消息服务才真正发送该消息；
3）对于那些未确认的消息或者取消的消息，需要有一个消息状态check系统定时去支付宝系统查询这个消息的状态并进行更新。
为什么需要这一步: 假设第2步支付宝扣款事务成功提交，接着系统挂了，会导致消息没有发送。

优点：消息数据独立存储，降低业务系统与消息系统间的耦合；
缺点：一次消息发送需要两次请求；业务服务需要实现消息状态查询接口。

 

解决消息重复投递问题
为什么相同的消息会被重复投递？重启后重发. 比如余额宝处理完消息msg后，发送了处理成功的消息给支付宝，正常情况下支付宝应该要删除消息msg，但如果支付宝这时候悲剧的挂了，重启后一看消息msg还在，就会继续发送消息msg。
解决方法很简单，在消息的消费者一方增加消息应用状态表（message_apply, 如下），每次来一个消息，在真正执行之前，先去消息应用状态表中查询一遍，如果找到说明是重复消息，丢弃即可，如果没找到才执行，同时插入到消息应用状态表

start transaction
  select count(*) as cnt from message_apply where msg_id=msg.msg_id;
  if cnt=0 then
    update B set amount=amount+10000 where userId=1;
    insert into message_apply(msg_id) values(msg.msg_id);
  end if;
commit;