分布式事务

innodb存储引擎，会在每行数据的最后加两个隐藏列，一个保存行的创建时间，一个保存行的删除时间，但是这儿存放的不是时间，而是事务id，事务id是mysql自己维护的自增的，全局唯一

在一个事务内查询的时候，mysql只会查询创建时间的事务id小于等于当前事务id的行

子主题

undo log多版本链条+ReadView机制

ReadView：基于undo log版本链条实现的一套试图机制，ReadView一旦生成就不会改变

RC隔离级别：的关键点在于，事务里每次查询都生成新的ReadView

RR隔离级别：的关键点在于，事务里每次查询都是同一个ReadView

public class ServiceB {

@Transactional(propagation = PROPAGATION_REQUIRES_NEW)
public void methodB() throws Exception {
// 一坨数据库操作
}

}

一般在单块系统开发，多人协作的时候比较常见，就是小A调用小B的模块，小A不管小B是成功还是不成功，自己都要提交，这个时候可以这么弄，就是说小B的操作不是构成小A的事务的重要组成部分，就是个分支

调用了invokeWithinTransaction()方法，核心的事务控制的逻辑都在invokeWithinTransaction()方法中，invokeWithinTransaction()方法其实是父类TransactionAspectSupport（Aspect这个名词就可以看出来了，这个东西一定是跟Spring AOP机制是有关系的，Aspect切面的意思，就是spring AOP的核心概念）

如果你的RefillDataCenterService.finishRefillData()方法报错了，抛了个异常出来，此时就会回滚事务

如果没有报错的话，就会提交你的事务，完成全部逻辑的执行

TM先发送个prepare消息给各个数据库，让各个库先把分布式事务里要执行的各种操作

各个数据库会准备好随时可以提交或者是回滚

然后各个数据库都返回一个响应消息给事务管理器，如果成功了就发送一个成功的消息，如果失败了就发送一个失败的消息

第一种情况，要是TM哥儿们发现某个数据库告诉他说，不好意思啊，我这儿失败了

把自己本地的那个事务回滚不就得了，然后各个库都回滚好了以后就通知TM，TM就认为整个分布式事务都回滚了

第二种情况，TM接收到所有的数据库返回的消息都是成功

提交好了通知下TM，TM要是发现所有数据库的事务都提交成功了

1、同步阻塞：在阶段一里执行prepare操作会占用资源，一直到整个分布式事务完成，才会释放资源

2、单点故障：TM是个单点，一旦挂掉就完蛋了

3、事务状态丢失：即使把TM做成一个双机热备的，一个TM挂了自动选举其他的TM出来，但是如果TM挂掉的同时，接收到commit消息的某个库也挂了，此时即使重新选举了其他的TM，压根儿不知道这个分布式事务当前的状态，因为不知道哪个库接收过commit消息，那个接收过commit消息的库也挂了

4、脑裂问题：在阶段二中，如果发生了脑裂问题，那么就会导致某些数据库没有接收到commit消息，有些库收到了commit消息，结果有些库没有收到

TM发送一个CanCommit消息给各个数据库，然后各个库返回个结果,不会执行实际的SQL语句的

就是各个库看看自己网络环境啊，各方面是否ready

会执行各个SQL语句，只是不提交

如果有个库对CanCommit消息返回了失败,TM发送abort消息给各个库,取消分布式事务

PreCommit阶段都返回了成功，那么发送DoCommit消息给各个库，提交事务，各个库如果都返回提交成功给TM，那么分布式事务成功

如果有个库对PreCommit返回的是失败，或者超时一直没返回，那么TM认为分布式事务失败，直接发abort消息给各个库，回滚，各个库回滚成功之后通知TM，分布式事务回滚成功

（1）引入了CanCommit阶段

（2）在DoCommit阶段，各个库自己也有超时机制,如果超时时间到了，还没收到TM发送的oCommit消息或者是abort消息，直接判定为TM可能出故障了,各个库就执行DoCommit操作，提交事务，解决了TM挂掉的单点问题

另外资源阻塞问题也能减轻一下，因为一个库如果一直接收不到DoCommit消息，不会一直锁着资源，人家自己会提交释放资源的，所有能减轻资源阻塞问题

也不是完美的，如果人家TM在DoCommit阶段发送了abort消息给各个库，结果因为脑裂问题，某个库没接收到abort消息，自己还因为超时机制的执行了commit操作

就是从ConnectionFactory里面去获取一个Connection，放到一个ConnectionPool里面去，这个池子里面放了数据库连接，在AtomikosDataSourceBean中，我们其实设置了DruidXADataSource

AtomikosDataSourceBean底层还是依赖于人家DruidXADataSource数据库连接池的

每个库都开启了XA START指令，都在依次通过prepareStatement()定义好每个库要执行的SQL，到这一步为止，其实所有的SQL都执行完了

同时对每个库都使用prepareStatement()方法定义好了每个库要执行哪些SQL语句，下一步就会去执行整个事务的这么一个提交，其实就会走2PC的一套逻辑了

如果说有人的prepare返回消息不是ok，那么就全量回滚，就是发送XA ROLLBACK指令，如果说所有人的返回消息都是ok，那么就对每个库发送XA COMMIT指令

这个rollback指令，其实同样会记录rollback指令的结果，如果rollback指令发送出去之后居然有某个库连rollback指令都没执行完成，那么就一旦走了XA分布式事务之后，性能极差

用了XA分布式事务之后，增加了100ms，耗时增加了15%，性能降低了15%，我可以大家说的一点是，如果在更加复杂的场景里面，XA分布式事务，性能最极端的情况下，会降低10倍

比如说你要是不用XA分布式事务，可能就是50ms，用了就是500ms

JTA + Atomikos实现XA分布式事务框架的架构设计的一个特点，基于Spring原生transaction事务控制的一个过程

JTA，其实是负责了Atomikos框架的一个指挥和调度，让他按照事务的基本步骤来走；Atomikos框架相当于是DTP模型里的TM，跟各个MySQL（RM）通信，XA START、XA END、XA COMMIT、XA ROLLBACK等符合XA规范的一套接口，调用

实现一套Atomikos框架，就是第一步，你需要跟JtaTransactionManager整合起来，需要JTA结合起来

创建分布式事务的时候，创建一个代表了分布式事务的对象；在各个SQL执行的时候，必须从你的DataSource里面获取Connection，对Connection的prepareStatement()方法的调用，你需要进行拦截，去对各个库执行XA START指令，以及定义好SQL；在提交事务的时候，你就需要去对各个库执行XA PREPARE指令，如果都成功，就执行XA COMMIT指令，如果失败，就执行XA ROLLBACK指令

比较有技术含量的一些点，还是动态代理的灵活的使用，自己去实现DataSource，Connection变成了你的动态代理

正常来说，账号1，amount = 9000，frozen = 1000；账号2，amount = 10000，frozen = 1000

账号1的try失败了，账号1的try操作的本地事务，回滚，导致账号1，amount = 10000，frozen = 0

账号2的try操作，压根儿就没机会去执行

正常来说，账号1，amount = 9000，frozen = 1000；账号2，amount = 10000，frozen = 1000

账号1的try操作先成功了，账号1，amount = 9000，frozen = 1000；账号2的try操作失败了，所以此时将frozen = 1000的操作就会回滚，然后amount = 10000，frozen = 0

bytetcc框架要自动感知到这个问题，然后将账号1执行cancel的逻辑，amount = 10000，frozen = 0

账号1，停留在try的那个阶段的数据，amount = 9000，frozen = 1000；账号2的confirm成功了，此时账号2变成amount = 11000，frozen = 0

明显的发现，bytetcc框架，并没有根据识别到的账号1的confirm的失败，来执行对应的cancel回滚的操作

bytetcc框架，就会不停的重试调用那个服务的confirm操作，直到他执行成功为止

账号1的confirm操作是成功的，amount = 9000，frozen = 0；账号2，amount = 10000，frozen = 1000

按照bytetcc框架的一个逻辑，肯定也是会账号1就那样放着吧，账号2会被不断的调用和重试，直到确保账号2的confirm操作成功

对外提供TCC服务的Controller必须加@Compensable注解(若没有实质业务, 也可以不必指定confirmableKey和cancellableKey)

若不加@Compensable注解, 则ByteTCC将其当成普通服务对待, 不接收Consumer端传播的事务上下文. 若它后续调用TCC服务, 则将开启新的TCC全局事务

@Compensable(interfaceClass = IAccountService.class, confirmableKey = "accountServiceConfirm", cancellableKey = "accountServiceCancel")
@RestController
public class AccountController implements IAccountService {

增加金额increaseAmount

减少金额decreaseAmount

provider执行的都是try操作

@Compensable(interfaceClass = ITransferService.class, confirmableKey = "transferServiceConfirm", cancellableKey = "transferServiceCancel")
@RestController
public class TransferController implements ITransferService {

@ResponseBody
@RequestMapping(value = "/transfer", method = RequestMethod.POST)
@Transactional
public void transfer(@RequestParam String sourceAcctId, @RequestParam String targetAcctId, @RequestParam double amount) {
this.acctService.decreaseAmount(sourceAcctId, amount);
this.increaseAmount(targetAcctId, amount);
}

@Service("transferServiceCancel")
public class TransferServiceCancel implements ITransferService {
@Autowired
private TransferDao transferDao;
@Transactional
public void transfer(String sourceAcctId, String targetAcctId, double amount) {
int value = this.transferDao.cancelIncrease(targetAcctId, amount);
if (value != 1) {
throw new IllegalStateException("ERROR!");
}
System.out.printf("exec decrease: acct= %s, amount= %7.2f%n", targetAcctId, amount);
}
}

@Service("transferServiceConfirm")
public class TransferServiceConfirm implements ITransferService {
@Autowired
private TransferDao transferDao;
@Transactional
public void transfer(String sourceAcctId, String targetAcctId, double amount) {
int value = this.transferDao.confirmIncrease(targetAcctId, amount);
if (value != 1) {
throw new IllegalStateException("ERROR!");
}
System.out.printf("done increase: acct= %s, amount= %7.2f%n", targetAcctId, amount);
}
}

@FeignClient(value = "SPRINGCLOUD-SAMPLE-PROVIDER")//调用PROVIDER接口
public interface IAccountService {
@RequestMapping(method = RequestMethod.POST, value = "/increase")
public void increaseAmount(@RequestParam("acctId") String accountId, @RequestParam("amount") double amount);
@RequestMapping(method = RequestMethod.POST, value = "/decrease")
public void decreaseAmount(@RequestParam("acctId") String accountId, @RequestParam("amount") double amount);
}

他是迎接一个http请求的第一道防线，他是一个interceptor，拦截器，他的本质跟TransactionInterceptor是一样的，就是在一个请求过来的时候，一定会先拦截掉

获取到了目标controller方法的@Transactional注解，还获取到了目标contorller的@Compensable注解，simplified（简化）模式，在当前的类中，就会有@CompensableConfirm和@CompensableCancel

按照调用顺序，如果你调用了一个服务，bytetcc分布式事务框架，会将你分布式事务中调用其他的服务作为一个resource放到一个resourceList中去，这样的话，分布式事务中调用了哪些其他服务，具体的状态是成功还是失败，都可以感知到

如果所有服务都try成功了，只要依次对这些服务调用confirm接口

如果一个分布式事务中要调用5个服务，结果只有2个服务调用成功了，第3个服务调用失败了，在进行cancel的时候，只要对try成功的服务执行cancel就可以了

基于RestTemplate组件调用每个已经成功try的服务接口，进行cancel

CompensableWork后台线程

timingRecover

this.fireNativeParticipantConfirm()

this.fireRemoteParticipantConfirm()

this.fireNativeParticipantCancel();

this.fireRemoteParticipantCancel()

最终，一定会通过SpringCloudCoordinator来发送http请求给账单服务的bytetcc controller，来执行/rollback接口，以执行账单服务自己本地的那个cancel逻辑

aggregate，聚合，说白了，就是一个组合

aggregate指的就是汽车 + 轮胎 + 引擎，三个东西聚合起来的一个整体

如果有个人要开车，那么只能调用汽车，因为汽车是aggregate root，他是aggregate对外提供的一个门面，不能有人直接调用轮胎或者是引擎吧

event sourcing的概念，其实说白了啊，就是事件回溯的意思

event sourcing模型里，很多时候也是会系统启动提前完成事件回溯，然后算出最新数据，放内存里缓存的，这样可以节约回溯开销，然后也可以多搞几个中间快照，每次从某个快照开始回溯

id name updated_age
1 张三 20
1 张三 1
1 张三 1
1 张三 1

Actor模型，这个模型的意思，就是说多线程在进行交互的时候，你假设一个Actor是一个线程，互相之间是基于消息进行通信的，每个Actor有个mailbox就是信箱，拿到了消息，然后内部单线程处理的

命令与查询职责的分离模型，大概意思，就是说，Command指的是类似增删改之类的处理命令，Query就是查询数据，然后让你在设计系统的时候，把增删改服务和查服务分离开来，就是在服务设计的层面，实现读写分离

C，command，指的就是增删改
Q，query，指的就是查询
你得用两个独立的服务来提供增删改操作和查询操作

写服务，负责接收增删改的请求，将数据更新到存储里面去，比如说可以使用mysql
就是会有专门的数据同步的组件，会将写服务的数据存储里面的数据，更新到读服务的数据存储里面去，比如说elasticsearch

读写分离的架构，最经典的，就是电商里的读写分离的场景，就是你的各种数据的变更统一都是走各种后台服务，都是mysql的

但是比如商品详情页，读流量那么的大，肯定不能直接从各个服务的mysql里查啊，那就私人了要，可以分离出来一个专门的读服务，每次数据变更就获取事件，同步到比如redis里，作为查数据的存储，然后别人就可以从读服务里查redis的数据

写服务在更新完自己的mysql之后，有一个专门同步组件会将数据同步到另外一个存储里去，比如es，或者是hbase，此时可以按照Event sourcing格式来同步数据，在es或者hbase里存放的是数据变更的快照，一个一个事件

读服务在读的时候，可以基于一个一个事件来聚合数据提供查询

专门帮助你实现CQRS模式，他里面的核心概念，都是DDD，领域驱动模型设计里的一些概念，包括了aggregate、event sourcing、actor、CQRS，那些概念就组成了一个CQRS框架，AxonFramework

在DDD里面提出了一个事务的模型，就是saga事务模型，AxonFramework里面实现的就是saga事务模型

CQRS

国外几乎都没什么人用的，不是很热门的框架，这个框架作者的更新速度很缓慢，也不是特别的好

其实是个CQRS框架，是基于DDD理论来的，你可以认为说，后面我们如果学习了DDD之后来建模和解决业务极度复杂的系统，那么就可以使用类似Axon这种框架了

好处就是他里面也提供了saga事务的支持和实现，而且这个框架总体来说名气大一些，活跃度高一些，github的star也多一些，1000多。。这还凑合是吧

BAT等大公司，其实主要都是tcc事务用来解决核心业务逻辑复杂调用链路里面的分布式事务问题

他不是一个saga事务框架，他其实是一个CQRS框架，只不过顺带支持了saga事务而已

没几个公司是用saga事务的，一般tcc + MQ最终一致性

（1）上游服务发送一个prepare消息，可以认为是一个待确认消息到RocketMQ

（2）RocketMQ会在自己内部保存这条消息，然后返回一个状态给上游服务

（3）接着上游服务执行本地事务

（4）如果失败了就发送rollback给RocketMQ，RocketMQ会删除掉那条消息，如果成功就发送commit给RocketMQ

（5）rocketmq会根据状态来处理消息，如果是rollback就删除消息，如果是commit就将消息标识为可以被下游服务消费

（6）下游服务消费消息

（7）接着下游服务会消费这个消息，执行本地事务

（8）下游服务然后返回ack给rocketmq，如果消费失败，或者是本地事务执行失败，或者ack发送失败，那么rocketmq都会有自己的重试策略，重发消息

（9）rocketmq如果能够成功的收到ack消息就会将消息删除

如果上游服务本地事务执行失败，也没发送rollback或者commit给rocketmq

下游服务只要自己保证了幂等性就可以了

有一种可能，在5这个步骤里，发送rabbitmq消息，看起来是成功了，其实消息并没有成功的投递到rabbitmq里去

本地的消息状态也变更为“已发送”，结果消息并没有投递到mq里去，下游的6789四个步骤没有执行

解决方案：可靠消息服务，他会发现已发送的但是超时的消息，重新再次投递一次消息到rabbitmq

可靠消息服务的机制设计的越是健壮，越是依赖你的可靠消息服务，而不是依赖MQ，就越好，最好对MQ的依赖是非常的轻量级的

如果你太过于依赖MQ了，可能导致MQ里的而一些bug，会导致你束手无策，无法解决

定义接口prepare，confirm，remove，finish