前言
虽然本人一直抱怨《微服务架构设计模式》中DDD模式下采用的Eventuate Tram Saga不算简单易用，但是为了更加深入了解原文作者的设计思路，还是花了点时间去阅读源码，并且为了自己日后自己返回来看的懂，就斗胆地对整个Eventuate Tram Saga从注册到执行的代码运行流程进行注释解读下，其中若是有什么错误疏漏以及需要改进的地方，希望各位在评论区指正。
源码讲解
1：Saga流程如何被记录注册
1-1 CreateOrderSaga类
//1-1构建流程，为了便以后续讲解，我们将一个step到下一个step之间的所有方法概括成一个“步骤”
public class CreateOrderSaga implements SimpleSaga {
private SagaDefinition sagaDefinition =
//初始化起点，调用1-2的step
step()
//调用1-3 的invokeLocal方法，this::reject是一个输入CreateOrderSagaData类型参数返回CommandWithDestination的方法
.invokeLocal(this::create)
.withCompensation(this::reject)
//步骤一
//调用1-2step
.step()
//调用1-3 的invokeParticipant方法
.invokeParticipant(this::reserveCredit)
.onReply(CustomerNotFound.class, this::handleCustomerNotFound)
.onReply(CustomerCreditLimitExceeded.class, this::handleCustomerCreditLimitExceeded)
//步骤二
//调用1-6step
.step()
.invokeLocal(this::approve)
.build();
｝
1-2 step初始化方法（saga流程的起点）
//1-2创建sagaDefinition的源码：因为实现了SimpleSaga，SimpleSaga默认方法
//此处的step对应1-1开头调用的step，1-1中后面的两个step并不调用此方法
default StepBuilder step() {
SimpleSagaDefinitionBuilder builder = new SimpleSagaDefinitionBuilder();
return new StepBuilder(builder);
}
1-3 StepBuilder类
//1-3StepBuilder内部方法：
public class StepBuilder{
private final SimpleSagaDefinitionBuilder parent;
//初始化时，这里parent被赋值为一个new SimpleSagaDefinitionBuilder;之后会用上第一次创建SimpleSagaDefinitionBuilder
public StepBuilder(SimpleSagaDefinitionBuilder builder) {
this.parent = builder;
}
//此处第一次时传入1-1中的this::create
public LocalStepBuilder invokeLocal(Consumer localFunction) {
return new LocalStepBuilder(parent, localFunction);
}
//此处传入1-1中的this::reserveCredit方法
public InvokeParticipantStepBuilder invokeParticipant(Function action) {
//调用1-6的withAction方法
return new InvokeParticipantStepBuilder(parent).withAction(Optional.empty(), action);
}
｝
1-4 LocalStepBuilder类
//1-4根据CreateOrderSaga 中的流程可得知，后续invokeLocal方法：
public class LocalStepBuilder {
private final SimpleSagaDefinitionBuilder parent;
private final Consumer localFunction;
private Optional> compensation = Optional.empty();
//设置父节点，设置执行方法
public LocalStepBuilder(SimpleSagaDefinitionBuilder parent, Consumer localFunction) {
this.parent = parent;
this.localFunction = localFunction;
}
//设置补偿方法，传入1-1中的this::reject
public LocalStepBuilder withCompensation(Consumer localCompensation) {
this.compensation = Optional.of(localCompensation);
return this;
}
//再次调用step方法，此处的step方法对应的是1-1中后面的两个step方法
public StepBuilder step() {
//调用1-5中的方法，将step之前接收的每个“步骤”放入到SimpleSagaDefinitionBuilder的List中
parent.addStep(makeLocalStep());
//创建一个新的“步骤”并承接之前的“步骤”
return new StepBuilder(parent);
}
//创建一个“步骤”
private LocalStep makeLocalStep() {
return new LocalStep(localFunction, compensation, localExceptionSavers, rollbackExceptions);
}
//对用1-1中的build方法，将最后一个“步骤”放入流程列表中，调用1-5中的方法创建一个SimpleSagaDefinition对象
public SagaDefinition build() {
//添加最后一个“步骤”并调用1-5中build()方法
parent.addStep(makeLocalStep());
return parent.build();
}
｝
1-5 SimpleSagaDefinitionBuilder类
//1-5
public class SimpleSagaDefinitionBuilder {
//存放“步骤”列表
private List> sagaSteps = new LinkedList();
//添加“步骤节点”到列表中
public void addStep(SagaStep sagaStep) {
sagaSteps.add(sagaStep);
}
//将所有步骤节点整合构建整个大“流程业务”
public SagaDefinition build() {
//将整个“步骤”节点列表传入2-3List steps内
return new SimpleSagaDefinition(sagaSteps);
}
}
1-6 InvokeParticipantStepBuilder类
//1-6
public class InvokeParticipantStepBuilder implements WithCompensationBuilder {
private Optional> action = Optional.empty();
public InvokeParticipantStepBuilder(SimpleSagaDefinitionBuilder parent) {
this.parent = parent;
}
//调用1-7的构造函数，1-1的this::reserveCredit方法传给ParticipantInvocationImpl构建出上面Optional> action
InvokeParticipantStepBuilder withAction(Optional> participantInvocationPredicate, Function action) {
this.action = Optional.of(new ParticipantInvocationImpl(participantInvocationPredicate, action));
return this;
}
public StepBuilder step() {
addStep();
return new StepBuilder(parent);
}
//本例中因为只使用了withAction方法，所以compensation, actionReplyHandlers, compensationReplyHandlers都为bull
//调用2-7的中ParticipantInvocationStep的构造方法，将action赋予participantInvocation
private void addStep() {
parent.addStep(new ParticipantInvocationStep(action, compensation, actionReplyHandlers, compensationReplyHandlers));
}
}
1-7 ParticipantInvocationImpl类
//1-7 AbstractParticipantInvocation实现ParticipantInvocation接口
public class ParticipantInvocationImpl extends AbstractParticipantInvocation {
private final boolean notification;
//commandBuilder对应的就是1-6中withAction方法action参数
private final Function commandBuilder;
public ParticipantInvocationImpl(Optional> invocablePredicate, Function commandBuilder) {
this(invocablePredicate, commandBuilder, false);
}
@Override
public boolean isSuccessfulReply(Message message) {
return CommandReplyOutcome.SUCCESS.name().equals(message.getRequiredHeader(ReplyMessageHeaders.REPLY_OUTCOME));
}
//被2-7调用，执行Function commandBuilder;并传入值为false的notification
@Override
public CommandWithDestinationAndType makeCommandToSend(Data data) {
return new CommandWithDestinationAndType(commandBuilder.apply(data), notification);
}
}
至此所有Saga步骤注册完整的业务流程完毕。
2、Saga实例化工厂如何运作
2-0 sagaInstanceFactory创建createOrderSaga
sagaInstanceFactory.create(createOrderSaga, data);
2-1 SagaInstanceFactory类
//2-1Saga实例化工厂
public class SagaInstanceFactory {
private Logger logger = LoggerFactory.getLogger(this.getClass());
private ConcurrentMap, SagaManager>> sagaManagers = new ConcurrentHashMap();
public SagaInstance create(Saga saga, SagaData data) {
SagaManager sagaManager = (SagaManager)sagaManagers.get(saga);
if (sagaManager == null)
throw new RuntimeException(("No SagaManager for " + saga));
//调用2-2的create方法
return sagaManager.create(data);
}
private SagaManager makeSagaManager(SagaManagerFactory sagaManagerFactory, Saga saga) {
SagaManagerImpl sagaManager = sagaManagerFactory.make(saga);
sagaManager.subscribeToReplyChannel();
return sagaManager;
}
}
2-2 SagaManagerImpl类
//2-2
public class SagaManagerImpl implements SagaManager {
@Override
public SagaInstance create(Data sagaData, Optional resource) {
SagaInstance sagaInstance = new SagaInstance(getSagaType(),null,"????",null,SagaDataSerde.serializeSagaData(sagaData), new HashSet());
//使用数据库保存saga实例
sagaInstanceRepository.save(sagaInstance);
String sagaId = sagaInstance.getId();
//此步骤没什么意义
saga.onStarting(sagaId, sagaData);
resource.ifPresent(r -> {
if (!sagaLockManager.claimLock(getSagaType(), sagaId, r)) {
throw new RuntimeException("Cannot claim lock for resource");
}
});
//getStateDefinition()获取SagaDefinition，即1-1中。
//调用2-4start方法，启动saga流程，构建第一个“步骤”的SagaActions
SagaActions actions = getStateDefinition().start(sagaData);
actions.getLocalException().ifPresent(e -> {
throw e;
});
//过程操作，传入参数：saga.getSagaType()是获取sagaData（操作数据）的类名，比CreateOrderSagaData并修改成固定格式
//sagaId:唯一sagaId
//sagaInstance:saga实例化
//sagaData:要操作的数据
//actions:第一个“步骤”的SagaActions
//调用下面的processActions方法
processActions(saga.getSagaType(), sagaId, sagaInstance, sagaData, actions);
return sagaInstance;
}
private void processActions(String sagaType, String sagaId, SagaInstance sagaInstance, Data sagaData, SagaActions actions) {
//进行循环
while (true) {
//如果传入的actions存在执行报错则信息执行if内的方法
if (actions.getLocalException().isPresent()) {
actions = getStateDefinition().handleReply(sagaType, sagaId, actions.getUpdatedState().get(), actions.getUpdatedSagaData().get(), MessageBuilder
.withPayload("{}")
.withHeader(ReplyMessageHeaders.REPLY_OUTCOME, CommandReplyOutcome.FAILURE.name())
.withHeader(ReplyMessageHeaders.REPLY_TYPE, Failure.class.getName())
.build());
} else {
// only do this if successful
//如果成功，通过消息队列发送给接收方。
//第一次进来时：因为1-1中构建的流程内，“步骤一”是调用本地方法，因此不需要发送消息
//第二次进来时：由于1-1中构建的流程内，“步骤二”是调用参与者方法，因此需要发送消息给参与者
String lastRequestId = sagaCommandProducer.sendCommands(this.getSagaType(), sagaId, actions.getCommands(), this.makeSagaReplyChannel());
//第一次进来时：lastRequestId第一个步骤时为null
//第二次进来时：返回一个请求Id
sagaInstance.setLastRequestId(lastRequestId);
//第一次进来时：更新“步骤一”sagaInstance实例状态信息：更新是否是最后节点（布尔值），更新是否需要补偿（布尔值），更新是否报错，更新更新的状态（对应2-5中executeStep方法的newState）
//第二次进来时：更新“步骤二”sagaInstance实例状态
updateState(sagaInstance, actions);
sagaInstance.setSerializedSagaData(SagaDataSerde.serializeSagaData(actions.getUpdatedSagaData().orElse(sagaData)));
//执行第一个步骤时，并不是最后一个步骤节点所以不进入if中
if (actions.isEndState()) {
performEndStateActions(sagaId, sagaInstance, actions.isCompensating(), actions.isFailed(), sagaData);
}
//使用数据库更新sagaInstance实例状态
sagaInstanceRepository.update(sagaInstance);
//public boolean isReplyExpected() {return (commands.isEmpty() || commands.stream().anyMatch(CommandWithDestinationAndType::isCommand)) && !local;}
//第一次进来时：在2-5的step.makeStepOutcome过程中因为将local设置为true，所以执行第一个步骤时actions.isReplyExpected()为false
if (actions.isReplyExpected()) {
break;
} else {
//模拟成功回复本地动作或通知，调用下面的simulateSuccessfulReplyToLocalActionOrNotification方法
//第一次进来时：传入“步骤一”的SagaActions，返回“步骤二”的SagaActions，继续循环
//第二次进来时：传入“步骤二”的SagaActions，返回“步骤三”的SagaActions
actions = simulateSuccessfulReplyToLocalActionOrNotification(sagaType, sagaId, actions);
}
}
}
}
//模拟成功回复本地动作或通知
private SagaActions simulateSuccessfulReplyToLocalActionOrNotification(String sagaType, String sagaId, SagaActions actions) {
//获取1-1中的整个业务流程后，调用2-4中的handleReply方法，并设置REPLY_OUTCOME和REPLY_TYPE头为success，最后返回“步骤二”SagaActions，重新进入上面的while循环中
return getStateDefinition().handleReply(sagaType, sagaId, actions.getUpdatedState().get(), actions.getUpdatedSagaData().get(), MessageBuilder
.withPayload("{}")
.withHeader(ReplyMessageHeaders.REPLY_OUTCOME, CommandReplyOutcome.SUCCESS.name())
.withHeader(ReplyMessageHeaders.REPLY_TYPE, Success.class.getName())
.build());
}
｝
2-3 AbstractSimpleSagaDefinition类
//2-3
public abstract class AbstractSimpleSagaDefinition,
ToExecute extends AbstractStepToExecute,
Provider extends AbstractSagaActionsProvider> {
protected Logger logger = LoggerFactory.getLogger(this.getClass());
//steps接受1-5中的sagaSteps
protected List steps;
public AbstractSimpleSagaDefinition(List steps) {
this.steps = steps;
}
//被2-4start方法调用
protected Provider firstStepToExecute(Data data) {
//SagaExecutionState.startingState()返回SagaExecutionState(-1, false);
//初始化时步骤节点currentlyExecuting为-1(因为0为第一个“步骤”，所以设置-1为起点)，是否补偿compensating为false
//开始执行下一个步骤节点（此处执行的是第一个“步骤”）
return nextStepToExecute(SagaExecutionState.startingState(), data);
}
//被2-4的handleReply方法调用
protected Provider sagaActionsForNextStep(String sagaType, String sagaId, Data sagaData, Message message,
SagaExecutionState state, Step currentStep, boolean compensating) {
//此处分两种情况:本地调方法用CommandReplyOutcome.SUCCESS.name().equals(message.getRequiredHeader(ReplyMessageHeaders.REPLY_OUTCOME));【例如1-1中的“步骤一”】；远程调用参与者方法用getParticipantInvocation(compensating).get().isSuccessfulReply(message);【例如1-1中的“步骤二”】
//由于“步骤一”是本地方法且头中包含信息为success，直接调用下方的nextStepToExecute方法
if (currentStep.isSuccessfulReply(compensating, message)) {
return nextStepToExecute(state, sagaData);
} else if (compensating) {
return handleFailedCompensatingTransaction(sagaType, sagaId, state, message);
} else {
return nextStepToExecute(state.startCompensating(), sagaData);
}
}
protected Provider nextStepToExecute(SagaExecutionState state, Data data) {
int skipped = 0;
//初始化compensating为false无需补偿
boolean compensating = state.isCompensating();
//初始化compensating为false，所以下个要执行的步骤节点+1，如果为true说明出错需要回滚，因此direction初始化为-1
int direction = compensating ? -1 : +1;
//第一次进入时:direction初始化为-1，所以i初始值为0，说明从第一个步骤开始执行；i必须小于节点长度；i根据direction来判断是否需要回滚到上一个节点还是进入下一个阶段
//第二次进入时:步骤一执行后state.getCurrentlyExecuting()变为0，所以i变为1，steps.get(i)获取“步骤二”。由于compensating依然为false，所以不进行回滚继续向下执行
for (int i = state.getCurrentlyExecuting() + direction; i >= 0 && i builder()
.withUpdatedState(SagaExecutionStateJsonSerde.encodeState(SagaExecutionState.makeFailedEndState()))
.withIsEndState(true)
.withIsCompensating(state.isCompensating())
.withIsFailed(true)
.build());
}
protected SagaActions makeEndStateSagaActions(SagaExecutionState state) {
return SagaActions.builder()
.withUpdatedState(SagaExecutionStateJsonSerde.encodeState(SagaExecutionState.makeEndState()))
.withIsEndState(true)
.withIsCompensating(state.isCompensating())
.build();
}
}
2-4 SimpleSagaDefinition类
//2-4
public class SimpleSagaDefinition{
public SimpleSagaDefinition(List> steps) {
super(steps);
}
@Override
public SagaActions start(Data sagaData) {
//执行2-3中的firstStepToExecute方法，启动第一个流程
return toSagaActions(firstStepToExecute(sagaData));
//构建SagaActions完毕，返回并继续2-2中的create方法内getStateDefinition().start(sagaData)后的代码
}
//被2-2中的processActions方法调用
@Override
public SagaActions handleReply(String sagaType, String sagaId, String currentState, Data sagaData, Message message) {
//将前一个“步骤”中的之前被JSON格式化（2-5的makeSagaActions把newState进行JSON格式化）的newState此处的currentState进行解码
SagaExecutionState state = SagaExecutionStateJsonSerde.decodeState(currentState);
//state.getCurrentlyExecuting()的值为“0”，因为初始化currentlyExecuting的计数是-1，而在2-5 中执行currentState.nextState(size())已经把当前“步骤”计数+1.而，所以step.get(0)获取了整个业务流程中的“步骤一”
SagaStep currentStep = steps.get(state.getCurrentlyExecuting());
//获取前一个“步骤”是否需要回滚
boolean compensating = state.isCompensating();
currentStep.getReplyHandler(message, compensating).ifPresent(handler -> invokeReplyHandler(message, sagaData, (d, m) -> {
handler.accept(d, m);
return null;
}));
//sagaActionsForNextStep会根据是否需要补偿来判断是采用nextStepToExecute（2-3）方法，还是调用handleFailedCompensatingTransaction（2-3）方法。
SagaActionsProvider sap = sagaActionsForNextStep(sagaType, sagaId, sagaData, message, state, currentStep, compensating);
return toSagaActions(sap);
}
//被2-3的nextStepToExecute中调用
@Override
protected StepToExecute makeStepToExecute(int skipped, boolean compensating, SagaStep step) {
return new StepToExecute(step, skipped, compensating);
}
//被2-3的nextStepToExecute中调用
@Override
protect服务器托管网ed SagaActionsProvider makeSagaActionsProvider(StepToExecute stepToExecute, Data data, SagaExecutionState state) {
//调用2-5的executeStep方法
return new SagaActionsProvider(() -> stepToExecute.executeStep(data, state));
}
}
2-5 executeStep方法
//2-5 StepToExecute类中的方法，被2-4中的makeSagaActionsProvider方法调用
public SagaActions executeStep(Data data, SagaExecutionState currentState) {
//nextState方法执行：SagaExecutionState(compensating ? currentlyExecuting - size : currentlyExecuting + size, compensating)
//protected int size() {return 1 + skipped;}
//需要补偿：则下一个状态为当前节点currentlyExecuting-已跳过（没有补偿事务）的节点的长度，即回到最后（有补偿事务）执行的步骤节点。
//不需要补偿：则下一个状态为当前节点currentlyExecuting+需要跳过（没有补偿事务）的节点的长度
//计算完后将currentlyExecuting进行更新
SagaExecutionState newState = currentState.nextState(size());
SagaActions.Builder builder = SagaActions.builder();
//当前正执行是否需要补偿回滚
boolean compensating = currentState.isCompensating();
//调用2-6中的makeStepOutcome方法
//第一次进来时：因为步骤一执行的是本地方法，调用的2-6中的makeStepOutcome，作用在于判断当前“步骤”是该执行补偿事务还是正常的本地事务，如果执行出现错误则返回一个带有报错信息的StepOutcome对象
//第二次进来时：因为步骤二执行的是让参与方执行方法，调用的2-7中的makeStepOutcomemakeStepOutcome
//执行StepOutcome的visit方法：将StepOutcome的RuntimeException类型的localOutcome属性赋值给SagaActions中的RuntimeException类型localException属性,同时将SagaActions中的local属性设置为ture;
step.makeStepOutcome(data, this.compensating).visit(builder::withIsLocal, builder::withCommands);
//SagaActions的makeSagaActions方法做两件事：将当前节点的数据newState格式化成JSON数据，newState信息包括：当前执行步骤的计数（第几个步骤）、是否回滚（布尔值）、是否是最后一个步骤（布尔值），是否出现错误（布尔值）。然后调用buildActions方法构建返回一个新的SagaActions
//String state = encodeState(newState);
//builder.buildActions(data, compensating, state, newState.isEndState());
//public SagaActions buildActions(Data data, boolean compensating, String state, boolean endState) {
//return withUpdatedSagaData(data)
// .withUpdatedState(state)
// .withIsEndState(endState)
// .withIsCompensating(compensating)
// .build();
//}
return makeSagaActions(builder, data, newState, compensating);
//第一次进来时：构建SagaActions完毕，返回到2-4中的start方法
//第二次进来时：构建SagagAcions完毕，返回到2-2中的simulateSuccessfulReplyToLocalActionOrNotification方法
}
2-6 LocalStep类
//2-6
public class LocalStep implements SagaStep {
private final Consumer localFunction;
private final Optional> compensation;
private final List> localExceptionSavers;
private final List> rollbackExceptions;
@Override
public StepOutcome makeStepOutcome(Data data, boolean compensating) {
try {
//如果需要回滚，执行回滚方法，compensation在1-1时的.withCompensation已经传入
if (compensating) {
//真正执行业务逻辑方法的地方
compensation.ifPresent(localStep -> localStep.accept(data));
} else {
//如果不需要回滚，直接执行补偿方法，localFunction在1-1的.invokeLocal时已经传入
localFunction.accept(data);
}
return makeLocalOutcome(Optional.empty());
} catch (RuntimeException e) {
localExceptionSavers.stream().filter(saver -> saver.shouldSave(e)).findFirst().ifPresent(saver -> saver.save(data, e));
if (rollbackExceptions.isEmpty() || rollbackExceptions.stream().anyMatch(c -> c.isInstance(e)))
return makeLocalOutcome(Optional.of(e));
else
throw e;
}
}
}
2-7 ParticipantInvocationStep类
//2-7
public class ParticipantInvocationStep implements SagaStep {
//participantInvocation被1-6的aaddStep方法传递action赋值
private Optional> participantInvocation;
private Optional> compensation;
public ParticipantInvocationStep(Optional> participantInvocation,
Optional> compensation,
Map> actionReplyHandlers,
Map> compensationReplyHandlers) {
this.actionReplyHandlers = actionReplyHandlers;
this.compensationReplyHandlers = compensationReplyHandlers;
this.服务器托管网participantInvocation = participantInvocation;
this.compensation = compensation;
}
//判断是否需要回滚，如果需要回滚执行compensation方法，如果不需要执行participantInvocation方法
private Optional> getParticipantInvocation(boolean compensating) {
return compensating ? compensation : participantInvocation;
}
@Override
public boolean isSuccessfulReply(boolean compensating, Message message) {
return getParticipantInvocation(compensating).get().isSuccessfulReply(message);
}
@Override
public StepOutcome makeStepOutcome(Data data, boolean compensating) {
//先调用getParticipantInvocation方法，此处假设不需要回滚所以返回participantInvocation
//调用makeRemoteStepOutcome方法，传入List commandsToSend 返回RemoteStepOutcome类型的结果
return StepOutcome.makeRemoteStepOutcome(getParticipantInvocation(compensating)
//调用1-7中的makeCommandToSend方法，执行消息发送方法
.map(pi -> pi.makeCommandToSend(data))
//将返回的CommandWithDestinationAndType包装成单元素列表
.map(Collections::singletonList)
//如果上述返回为空则返回一个空列表
.orElseGet(Collections::emptyList));
}
}
图解流程
有空再补….
简单案例
花了两天写了个以领域驱动为思想的Saga模式事务管理简陋框架，主要为了讲解：领域驱动模型DDD（三）——使用Saga管理事务 教学而设计的，目前只能在单体架构中使用，后续有时间会更新分布式情况下的新版本。请记住，领域驱动模型是一种思想，它不一定捆绑分布式微服务，只是领域驱动模型思想更有利于分布式情况下对微服务应用的划分。
项目框架地址：https://github.com/CG-Lin/mvn-lin
服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net
机房租用，北京机房租用，IDC机房托管， http://www.fwqtg.net
相关推荐: GaussDB数据库SQL系列-SQL与ETL浅谈
尊嘟假嘟，成都程序员也 996 吗？ 目录 一、前言 二、SQL与ETL的概述 三、ETL过程中的SQL示例（GaussDB） 1、提取（Extract） 2、转换（Transform） 3、加载（Load） 四、附DataArts Studio介绍 五、小结…
