TX-LCN核心源码解读
TX-LCN是基于Java编写的分布式事务解决方案框架，主要提供三种主流的解决方案
- LCN模式，通过代理JDBC Connection来控制协调多组本地原子事务的提交与关闭
- TCC模式，属于框架级别解决方案，对业务入侵性极大
- TXC模式，核心方案为查询 + 分布式锁的分布式事务解决方案，由淘宝团队提出
核心组件
- TC，作为分布式事务组件的客户端角色，主要作用在于治理本地事务
- TM，作为分布式事务组件的服务端角色，主要作用在于协调事务组
核心概念
- 事务组，
group，描述整个分布式环境下运行的各个事务（以一个request所需要完成的事务）组合而成的一组事务。 - 事务单元，
unit，描述一个事务组内除开主事务之外的从事务，一个从事务表示一个事务单元。
核心流程
事务核心流程.png
核心源码
以LCN模式作为实践方案，源码也以LCN 作为解读对象。LCN 模式是什么模式？
LCN模式是通过代理Connection的方式实现对本地事务的操作，然后在由TxManager统一协调控制事务。当本地事务提交回滚或者关闭连接时将会执行假操作，该代理的连接将由LCN连接池管理。
创建事务组
/**
* Client创建事务组操作集合
*
* @param groupId groupId
* @param unitId unitId
* @param transactionInfo transactionInfo
* @param transactionType transactionType
* @throws TransactionException 创建group失败时抛出
*/
public void createGroup(String groupId, String unitId, TransactionInfo transactionInfo, String transactionType)
throws TransactionException {
//创建事务组
try {
// 日志
txLogger.transactionInfo(groupId, unitId,
"create group > {} > groupId: {xid}, unitId: {uid}", transactionType);
// 创建事务组消息
reliableMessenger.createGroup(groupId);
// 缓存发起方切面信息
aspectLogger.trace(groupId, unitId, transactionInfo);
} catch (RpcException e) {
// 通讯异常
dtxExceptionHandler.handleCreateGroupMessageException(groupId, e);
} catch (LcnBusinessException e) {
// 创建事务组业务失败
dtxExceptionHandler.handleCreateGroupBusinessException(groupId, e.getCause());
}
txLogger.transactionInfo(groupId, unitId, "create group over");
}
TC 发起创建事务组仅仅是像服务端发起请求，参数为groupId ，核心的代码在于TM 接到请求后的处理。
public class CreateGroupExecuteService implements RpcExecuteService {
// ...
@Override
public Serializable execute(TransactionCmd transactionCmd) throws TxManagerException {
// ...
transactionManager.begin(transactionCmd.getGroupId());
}
}
public class DefaultDTXContextRegistry implements DTXContextRegistry {
private final FastStorage fastStorage;
@Override
public DTXContext create(String groupId) throws TransactionException {
// ..
fastStorage.initGroup(groupId);
}
}
public class RedisStorage implements FastStorage {
// ...
@Override
public void initGroup(String groupId) {
// 将groupId存入Redis
redisTemplate.opsForHash().put(REDIS_GROUP_PREFIX + groupId, "root", "");
redisTemplate.expire(REDIS_GROUP_PREFIX + groupId, managerConfig.getDtxTime() + 10000, TimeUnit.MILLISECONDS);
}
}
加入事务组
/**
* Client加入事务组操作集合
*
* @param groupId groupId
* @param unitId unitId
* @param transactionType transactionType
* @param transactionInfo transactionInfo
* @throws TransactionException 加入事务组失败时抛出
*/
public void joinGroup(String groupId, String unitId, String transactionType, TransactionInfo transactionInfo) throws TransactionException {
// 询问TM加入事务组
// 该groupId由远程RPC通过header方式携带到从事务
reliableMessenger.joinGroup(groupId, unitId, transactionType, DTXLocalContext.transactionState());
// 异步检测
dtxChecking.startDelayCheckingAsync(groupId, unitId, transactionType);
// ...
}
//
@Override
public void startDelayCheckingAsync(String groupId, String unitId, String transactionType) {
txLogger.taskInfo(groupId, unitId, "start delay checking task");
// 异步阻塞的方式
ScheduledFuture scheduledFuture = scheduledExecutorService.schedule(() -> {
try {
TxContext txContext = globalContext.txContext(groupId);
if (Objects.nonNull(txContext)) {
synchronized (txContext.getLock()) {
txLogger.info(groupId, unitId, Transactions.TAG_TASK,
"checking waiting for business code finish.");
txContext.getLock().wait();
}
}
int state = reliableMessenger.askTransactionState(groupId, unitId);// 询问事务组是否成功
txLogger.taskInfo(groupId, unitId, "ask transaction state {}", state);
if (state == -1) {
txLogger.error(this.getClass().getSimpleName(), "delay clean transaction error.");
onAskTransactionStateException(groupId, unitId, transactionType);
} else {
transactionCleanTemplate.clean(groupId, unitId, transactionType, state);// 事务清理，即commit or rollback
aspectLogger.clearLog(groupId, unitId);
}
} catch (RpcException e) {
onAskTransactionStateException(groupId, unitId, transactionType);
} catch (TransactionClearException | InterruptedException e) {
txLogger.error(this.getClass().getSimpleName(), "{} clean transaction error.", transactionType);
}
}, clientConfig.getDtxTime(), TimeUnit.MILLISECONDS);// 时间为最大事务时间，该时间由TM配置，在TC初始化时从TM放取到
delayTasks.put(groupId + unitId, scheduledFuture);
}
// 询问事务是否成功
private void onAskTransactionStateException(String groupId, String unitId, String transactionType) {
try {
// 通知TxManager事务补偿
txMangerReporter.reportTransactionState(groupId, unitId, TxExceptionParams.ASK_ERROR, 0);
log.warn("{} > has compensation info!", transactionType);
// 事务回滚, 保留适当的补偿信息
transactionCleanTemplate.compensationClean(groupId, unitId, transactionType, 0);
} catch (TransactionClearException e) {
log.error("{} > clean transaction error.", transactionType);
}
}
TC 核心工作是申请加入事务组，并启动异步任务在事务最大时间后访问TM 事务组的全局事务状态来进行事务协调。
创建异步任务对象并将其缓存在本地内存的delayTasks， 如果在事务最大时间内已经完成并调用则将该任务取消。
@Override
public void stopDelayChecking(String groupId, String unitId) {
ScheduledFuture scheduledFuture = delayTasks.get(groupId + unitId);
if (Objects.nonNull(scheduledFuture)) {
txLogger.taskInfo(groupId, unitId, "cancel {}:{} checking.", groupId, unitId);
scheduledFuture.cancel(true); // 取消任务
}
}
而在TM 端则进行如下处理
// 加入事务组
// RedisStorage
public void saveTransactionUnitToGroup(String groupId, TransactionUnit transactionUnit) throws FastStorageException {
if (Optional.ofNullable(redisTemplate.hasKey(REDIS_GROUP_PREFIX + groupId)).orElse(false)) {
redisTemplate.opsForHash().put(REDIS_GROUP_PREFIX + groupId, transactionUnit.getUnitId(), transactionUnit);
return;
}
throw new FastStorageException("attempts to the non-existent transaction group " + groupId, FastStorageException.EX_CODE_NON_GROUP);
}
// TC 询问该事务组全局事务状态
public int transactionState(String groupId) {
int state = exceptionService.transactionState(groupId); // 查询数据库t_tx_exception得到该事务的状态
//存在数据时返回数据状态
if (state != -1) {
return state;
}
// 查询redis该事务组的全局状态
return dtxContextRegistry.transactionState(groupId);
// 为什么会先查数据库，再查redis ？ 后文会有说明，主要是考虑事务补偿时的问题
}
// 返回的最终状态
public class AskTransactionStateExecuteService implements RpcExecuteService {
@Override
public Serializable execute(TransactionCmd transactionCmd) {
int state = transactionManager.transactionState(transactionCmd.getGroupId());
return state == -1 ? 0 : state;
}
}
通知事务组
主事务的业务代码执行完毕，最终必须调用通知事务组进行全局事务协调。通知完成后进行事务清理
/**
* Client通知事务组操作集合
*
* @param groupId groupId
* @param unitId unitId
* @param transactionType transactionType
* @param state transactionState
*/
public void notifyGroup(String groupId, String unitId, String transactionType, int state) {
// ...
// 事务通知
reliableMessenger.notifyGroup(groupId, state);
// 事务清理
transactionCleanTemplate.clean(groupId, unitId, transactionType, state);
// 通知异常（RPC调用异常）
dtxExceptionHandler.handleNotifyGroupMessageException(Arrays.asList(groupId, state, unitId, transactionType), e);
// ...
}
// 当TC调用TM抛出异常时，会正常的按照当前事务的状态进行提交，并将结果上报到TM
public void handleNotifyGroupMessageException(Object params, Throwable ex) {
// 参数中取出事务的状态
// ....
// 按状态正常结束事务（切面补偿记录将保留）
// TxManager 存在请求异常或者响应异常两种情况。当请求异常时这里的业务需要补偿，当响应异常的时候需要做状态的事务清理。
// 请求异常时
// 参与放会根据上报补偿记录做事务的提交。
// 响应异常时
// 参与反会正常提交事务，本地业务提示事务。
// 该两种情况下补偿信息均可以忽略,可直接把本地补偿记录数据删除。
String unitId = (String) paramList.get(2);
String transactionType = (String) paramList.get(3);
try {
transactionCleanTemplate.compensationClean(groupId, unitId, transactionType, state);// 本地事务提交
} catch (TransactionClearException e) {
log.error("{} > compensationClean transaction error.", transactionType);
}
// 上报Manager，上报直到成功.
txMangerReporter.reportTransactionState(groupId, null, TxExceptionParams.NOTIFY_GROUP_ERROR, state);
// 提交的事务记录到t_tx_exception表中，所以会看到前文TC询问事务状态时，会优先查询数据库，而不是直接查redis
}
通知事务组的概念，应该理解为，主事务告知TM 进行全部的事务协调，即TM 仅会通知各个从事务进行commit or rollback，而不会通知主事务进行commit or rollback 。因为在前文看到创建事务组时，TM 并没有将主事务unitId 记录下来。而从事务加入事务组时，除了记录全局事务组Id，还包括事务单元unitId .
public Serializable execute(TransactionCmd transactionCmd) throws TxManagerException {
try {
// 从redis取事务状态
int transactionState = transactionManager.transactionStateFromFastStorage(transactionCmd.getGroupId());
boolean hasThrow = false;
if (transactionState == 0) {
commitState = 0;
hasThrow = true;
}
// 事务状态为1进行全局事务提交
if (commitState == 1) {
transactionManager.commit(dtxContext);
} else if (commitState == 0) {
transactionManager.rollback(dtxContext);
}
// ...
} catch (TransactionException e) {
throw new TxManagerException(e);
} finally {
transactionManager.close(transactionCmd.getGroupId());
// 系统日志
txLogger.transactionInfo(transactionCmd.getGroupId(), "", "notify group over");
}
return null;
}
// 事务通知
private void notifyTransaction(DTXContext dtxContext, int transactionState) throws TransactionException {
for (TransactionUnit transUnit : dtxContext.transactionUnits()) {
NotifyUnitParams notifyUnitParams = new NotifyUnitParams();
notifyUnitParams.setGroupId(dtxContext.getGroupId());
notifyUnitParams.setUnitId(transUnit.getUnitId());
notifyUnitParams.setUnitType(transUnit.getUnitType());
notifyUnitParams.setState(transactionState);
txLogger.info(dtxContext.getGroupId(),
notifyUnitParams.getUnitId(), Transactions.TAG_TRANSACTION, "notify %s's unit: %s",
transUnit.getModId(), transUnit.getUnitId());
try {
// 这里在5.0.1会出现信道问题，什么是信道问题？比如此时有两台push注册到TM上，而某一刻的全局事务所在的本地事务只在其中一台，而通知的时候如果modId一致则会取到第一个
// 如下get(0) . 解决的办法是生成modId时去的是Mac地址+端口+服务名称，保证了不同实例的全局唯一
List<String> modChannelKeys = rpcClient.remoteKeys(transUnit.getModId());
if (modChannelKeys.isEmpty()) {
// record exception
throw new RpcException("offline mod.");
}
MessageDto respMsg =
rpcClient.request(modChannelKeys.get(0), MessageCreator.notifyUnit(notifyUnitParams));
if (!MessageUtils.statusOk(respMsg)) {
// 提交/回滚失败的消息处理
List<Object> params = Arrays.asList(notifyUnitParams, transUnit.getModId());
rpcExceptionHandler.handleNotifyUnitBusinessException(params, respMsg.loadBean(Throwable.class));
}
} catch (RpcException e) {
// 提交/回滚通讯失败
List<Object> params = Arrays.asList(notifyUnitParams, transUnit.getModId());
rpcExceptionHandler.handleNotifyUnitMessageException(params, e);
} finally {
txLogger.transactionInfo(dtxContext.getGroupId(), notifyUnitParams.getUnitId(), "notify unit over");
}
}
}
// 当通知出现异常时，将信息记录到t_tx_exception表
public class ManagerRpcExceptionHandler implements RpcExceptionHandler {
@Override
public void handleNotifyUnitMessageException(Object params, Throwable e) {
// notify unit message error, write txEx
List paramList = ((List) params);
String modName = (String) paramList.get(1);
NotifyUnitParams notifyUnitParams = (NotifyUnitParams) paramList.get(0);
WriteTxExceptionDTO writeTxExceptionReq = new WriteTxExceptionDTO(notifyUnitParams.getGroupId(),
notifyUnitParams.getUnitId(), modName, notifyUnitParams.getState());
writeTxExceptionReq.setRegistrar((short) 0);
compensationService.writeTxException(writeTxExceptionReq);// 记录到t_tx_exception
// 记住客户端主动查询时，优先查数据库，再查redis的事务状态
}
}
总结
TX-LCN作为分布式解决方案是比较优秀的方案，代码逻辑也比较简单，但是如果应用Crash，就可能出现数据不一致的情况，而且这种数据不一致的情况必须人肉修复。
比如主事务在进行NotifyGroup 时出现RpcException 主事务会根据当前事务的状态进行commit or rollback ，之后会上报TM 记录补偿信息，假如在记录补偿时失败了（应用在这个点Crash）了，那么主事务提交了，并且TM 并不能完整地协调好从事务 的全局事务状态。
为什么需要人肉修复呢？其实从源码上可以分析出，TX-LCN解决的场景时将本地的事务通过事务协调器进行协调，但是本质上并没有将事务分布式节点化，即本地事务的成功与失败无法在不同的节点进行处理。
