分布式事务涉及多个微服务，其中每个服务执行事务的一部分。例如，电子商务平台可能涉及支付、库存和订单管理等服务。这些服务需要协同工作才能完成交易。但是，如果其中一项服务失败会发生什么情况？
想象一个电子商务应用程序，在下订单期间会发生以下步骤：
如果在扣款之后、创建订单之前库存服务失败，系统最终会处于不一致的状态。客户已付费，但未下订单。
为了处理此类故障，可以考虑使用具有两阶段提交协议的分布式事务。然而，这引入了几个问题：
在分布式系统中，事务通常跨越多个微服务。确保所有服务要么成功完成，要么根本没有完成是具有挑战性的。处理此问题的传统方法（使用具有两阶段提交的分布式事务）可能会因高延迟、紧密耦合和可用性降低等问题而出现问题。
Saga 模式提供了一种更灵活的方法。 Saga 模式不是尝试将事务作为单个单元执行，而是将事务分解为可以独立执行的更小的、孤立的步骤。每个步骤都是一个本地事务，更新数据库，然后触发下一步。如果某个步骤失败，系统会执行补偿操作以撤消先前步骤所做的更改，确保系统可以返回到一致的状态。
Saga 模式本质上是一系列依次执行的较小事务。其工作原理如下：
实现Saga模式主要有两种方式：编排和编排。
在编舞传奇中，没有中央协调员。相反，Saga 中涉及的每个服务都会侦听事件并根据先前步骤的结果决定何时采取行动。这种方法是分散的，允许服务独立运行。其工作原理如下：
编排的优点：
编舞的挑战：
在 Orchestration Saga 中，中央编排器控制事务流。协调器确定步骤的顺序并处理服务之间的通信。其工作原理如下：
编排的优点：
编排的挑战：
让我们考虑电子商务场景并使用Saga模式来实现它。
在我们的咖啡购买场景中，每项服务都代表一次本地交易。咖啡服务充当这个传奇的协调者，协调其他服务来完成购买。
以下是这个传奇如何运作的详细说明：
在我的传奇实现中，每个 SagaItemBuilder 代表我们分布式事务流中的一个步骤。 ActionBuilder 定义了要执行的操作，包括主操作和发生错误时执行的回滚操作。 ActionBuilder 封装了三部分信息：
component ：要调用的方法所在的 Bean 实例。
method ：要调用的方法的名称。
args ：要传递给方法的参数。
ActionBuilder
public class ActionBuilder {
private Object component;
private String method;
private Object[] args;
public static ActionBuilder builder() {
return new ActionBuilder();
}
public ActionBuilder component(Object component) {
this.component = component;
return this;
}
public ActionBuilder method(String method) {
this.method = method;
return this;
}
public ActionBuilder args(Object... args) {
this.args = args;
return this;
}
public Object getComponent() { return component; }
public String getMethod() { return method; }
public Object[] getArgs() { return args; }
}
SagaItemBuilder
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
public class SagaItemBuilder {
private ActionBuilder action;
private Map<Class<? extends Exception>, ActionBuilder> onBehaviour;
public static SagaItemBuilder builder() {
return new SagaItemBuilder();
}
public SagaItemBuilder action(ActionBuilder action) {
this.action = action;
return this;
}
public SagaItemBuilder onBehaviour(Class<? extends Exception> exception, ActionBuilder action) {
if (Objects.isNull(onBehaviour)) onBehaviour = new HashMap<>();
onBehaviour.put(exception, action);
return this;
}
public ActionBuilder getAction() {
return action;
}
public Map<Class<? extends Exception>, ActionBuilder> getBehaviour() {
return onBehaviour;
}
}
场景
import java.util.ArrayList;
import java.util.List;
public class Scenarios {
List<SagaItemBuilder> scenarios;
public static Scenarios builder() {
return new Scenarios();
}
public Scenarios scenario(SagaItemBuilder sagaItemBuilder) {
if (scenarios == null) scenarios = new ArrayList<>();
scenarios.add(sagaItemBuilder);
return this;
}
public List<SagaItemBuilder> getScenario() {
return scenarios;
}
}
以下是我如何提交分发事务。
package com.example.demo.saga;
import com.example.demo.saga.exception.CanNotRollbackException;
import com.example.demo.saga.exception.RollBackException;
import com.example.demo.saga.pojo.ActionBuilder;
import com.example.demo.saga.pojo.SagaItemBuilder;
import com.example.demo.saga.pojo.Scenarios;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.Set;
@Component
public class DTC {
public boolean commit(Scenarios scenarios) throws Exception {
validate(scenarios);
for (int i = 0; i < scenarios.getScenario().size(); i++) {
SagaItemBuilder scenario = scenarios.getScenario().get(i);
ActionBuilder action = scenario.getAction();
Object bean = action.getComponent();
String method = action.getMethod();
Object[] args = action.getArgs();
try {
invoke(bean, method, args);
} catch (Exception e) {
rollback(scenarios, i, e);
return false;
}
}
return true;
}
private void rollback(Scenarios scenarios, Integer failStep, Exception currentStepFailException) {
for (int i = failStep; i >= 0; i--) {
SagaItemBuilder scenario = scenarios.getScenario().get(i);
Map<Class<? extends Exception>, ActionBuilder> behaviours = scenario.getBehaviour();
Set<Class<? extends Exception>> exceptions = behaviours.keySet();
ActionBuilder actionWhenException = null;
if (failStep == i) {
for(Class<? extends Exception> exception: exceptions) {
if (exception.isInstance(currentStepFailException)) {
actionWhenException = behaviours.get(exception);
}
}
if (actionWhenException == null) actionWhenException = behaviours.get(RollBackException.class);
} else {
actionWhenException = behaviours.get(RollBackException.class);
}
Object bean = actionWhenException.getComponent();
String method = actionWhenException.getMethod();
Object[] args = actionWhenException.getArgs();
try {
invoke(bean, method, args);
} catch (Exception e) {
throw new CanNotRollbackException("Error in %s belong to %s. Can not rollback transaction".formatted(method, bean.getClass()));
}
}
}
private void validate(Scenarios scenarios) throws Exception {
for (int i = 0; i < scenarios.getScenario().size(); i++) {
SagaItemBuilder scenario = scenarios.getScenario().get(i);
ActionBuilder action = scenario.getAction();
if (action.getComponent() == null) throw new Exception("Missing bean in scenario");
if (action.getMethod() == null) throw new Exception("Missing method in scenario");
Map<Class<? extends Exception>, ActionBuilder> behaviours = scenario.getBehaviour();
Set<Class<? extends Exception>> exceptions = behaviours.keySet();
if (exceptions.contains(null)) throw new Exception("Exception can not be null in scenario has method %s, bean %s " .formatted(action.getMethod(), action.getComponent().getClass()));
if (!exceptions.contains(RollBackException.class)) throw new Exception("Missing default RollBackException in scenario has method %s, bean %s " .formatted(action.getMethod(), action.getComponent().getClass()));
}
}
public String invoke(Object bean, String methodName, Object... args) throws Exception {
try {
Class<?>[] paramTypes = new Class[args.length];
for (int i = 0; i < args.length; i++) {
paramTypes[i] = parameterType(args[i]);
}
Method method = bean.getClass().getDeclaredMethod(methodName, paramTypes);
Object result = method.invoke(bean, args);
return result != null ? result.toString() : null;
} catch (Exception e) {
throw e;
}
}
private static Class<?> parameterType (Object o) {
if (o instanceof Integer) {
return int.class;
} else if (o instanceof Boolean) {
return boolean.class;
} else if (o instanceof Double) {
return double.class;
} else if (o instanceof Float) {
return float.class;
} else if (o instanceof Long) {
return long.class;
} else if (o instanceof Short) {
return short.class;
} else if (o instanceof Byte) {
return byte.class;
} else if (o instanceof Character) {
return char.class;
} else {
return o.getClass();
}
}
}
我有 3 个调用外部服务的服务：BillingService、OrderService、PaymentService。
订单服务
package com.example.demo.service;
import org.springframework.stereotype.Service;
@Service
public class OrderService {
public String prepareOrder(String name, int number) {
System.out.println("Prepare order for %s with order id %d ".formatted(name, number));
return "Prepare order for %s with order id %d ".formatted(name, number);
}
public void Rollback_prepareOrder_NullPointException() {
System.out.println("Rollback prepareOrder because NullPointException");
}
public void Rollback_prepareOrder_RollBackException() {
System.out.println("Rollback prepareOrder because RollBackException");
}
}
计费服务
package com.example.demo.service;
import org.springframework.stereotype.Service;
@Service
public class BillingService {
public String prepareBilling(String name, int number) {
System.out.println("Prepare billing for %s with order id %d ".formatted(name, number));
return "Prepare billing for %s with order id %d ".formatted(name, number);
}
public String createBilling(String name, int number) {
System.out.println("Create billing for %s with order id %d ".formatted(name, number));
return "Create billing for %s with order id %d ".formatted(name, number);
}
public void Rollback_prepareBilling_NullPointException() {
System.out.println("Rollback prepareBilling because NullPointException");
}
public void Rollback_prepareBilling_ArrayIndexOutOfBoundsException() {
System.out.println("Rollback prepareBilling because ArrayIndexOutOfBoundsException");
}
public void Rollback_prepareBilling_RollBackException() {
System.out.println("Rollback prepareBilling because RollBackException");
}
public void Rollback_createBilling_NullPointException() {
System.out.println("Rollback createBilling because NullPointException");
}
public void Rollback_createBilling_ArrayIndexOutOfBoundsException() {
System.out.println("Rollback createBilling because ArrayIndexOutOfBoundsException");
}
public void Rollback_createBilling_RollBackException() {
System.out.println("Rollback createBilling because RollBackException");
}
}
付款服务
package com.example.demo.service;
import org.springframework.stereotype.Service;
@Service
public class PaymentService {
public String createPayment() {
System.out.println("Create payment");
return "Create payment";
}
public void Rollback_createPayment_NullPointException() {
System.out.println("Rollback createPayment because NullPointException");
}
public void Rollback_createPayment_RollBackException() {
System.out.println("Rollback createPayment because RollBackException");
}
}
在 Coffee Service 中，我按如下方式实现它，创建一个场景，然后提交它。
package com.example.demo.service;
import com.example.demo.saga.DTC;
import com.example.demo.saga.exception.RollBackException;
import com.example.demo.saga.pojo.ActionBuilder;
import com.example.demo.saga.pojo.SagaItemBuilder;
import com.example.demo.saga.pojo.Scenarios;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class CoffeeService {
@Autowired
private OrderService orderService;
@Autowired
private BillingService billingService;
@Autowired
private PaymentService paymentService;
@Autowired
private DTC dtc;
public String test() throws Exception {
Scenarios scenarios = Scenarios.builder()
.scenario(
SagaItemBuilder.builder()
.action(ActionBuilder.builder().component(orderService).method("prepareOrder").args("tuanh.net", 123))
.onBehaviour(NullPointerException.class, ActionBuilder.builder().component(orderService).method("Rollback_prepareOrder_NullPointException").args())
.onBehaviour(RollBackException.class, ActionBuilder.builder().component(orderService).method("Rollback_prepareOrder_RollBackException").args())
).scenario(
SagaItemBuilder.builder()
.action(ActionBuilder.builder().component(billingService).method("prepareBilling").args("tuanh.net", 123))
.onBehaviour(NullPointerException.class, ActionBuilder.builder().component(billingService).method("Rollback_prepareBilling_NullPointException").args())
.onBehaviour(RollBackException.class, ActionBuilder.builder().component(billingService).method("Rollback_prepareBilling_RollBackException").args())
).scenario(
SagaItemBuilder.builder()
.action(ActionBuilder.builder().component(billingService).method("createBilling").args("tuanh.net", 123))
.onBehaviour(NullPointerException.class, ActionBuilder.builder().component(billingService).method("Rollback_createBilling_ArrayIndexOutOfBoundsException").args())
.onBehaviour(RollBackException.class, ActionBuilder.builder().component(billingService).method("Rollback_createBilling_RollBackException").args())
).scenario(
SagaItemBuilder.builder()
.action(ActionBuilder.builder().component(paymentService).method("createPayment").args())
.onBehaviour(NullPointerException.class, ActionBuilder.builder().component(paymentService).method("Rollback_createPayment_NullPointException").args())
.onBehaviour(RollBackException.class, ActionBuilder.builder().component(paymentService).method("Rollback_createPayment_RollBackException").args())
);
dtc.commit(scenarios);
return "ok";
}
}
当我在创建账单时例外。
public String createBilling(String name, int number) {
throw new NullPointerException();
}
结果
2024-08-24T14:21:45.445+07:00 INFO 19736 --- [demo] [main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port 8080 (http) with context path '/' 2024-08-24T14:21:45.450+07:00 INFO 19736 --- [demo] [main] com.example.demo.DemoApplication : Started DemoApplication in 1.052 seconds (process running for 1.498) 2024-08-24T14:21:47.756+07:00 INFO 19736 --- [demo] [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring DispatcherServlet 'dispatcherServlet' 2024-08-24T14:21:47.756+07:00 INFO 19736 --- [demo] [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet : Initializing Servlet 'dispatcherServlet' 2024-08-24T14:21:47.757+07:00 INFO 19736 --- [demo] [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet : Completed initialization in 1 ms Prepare order for tuanh.net with order id 123 Prepare billing for tuanh.net with order id 123 Rollback createBilling because RollBackException Rollback prepareBilling because RollBackException Rollback prepareOrder because RollBackException
查看我的 GitHub 存储库
总之，Saga 模式通过将分布式事务分解为更小的、可管理的步骤，为管理分布式事务提供了一个强大的解决方案。编排和编排之间的选择取决于系统的特定需求和架构。编排提供松散耦合和弹性，而编排提供集中控制和更轻松的监控。通过使用 Saga 模式仔细设计系统，您可以在分布式微服务架构中实现一致性、可用性和灵活性。
如果您对在系统中实现 Saga 模式有任何疑问或需要进一步说明，请随时在下面发表评论！
阅读更多帖子：Saga 模式如何解决分布式事务问题：方法和实际示例
以上是Saga 模式如何解决分布式事务问题：方法和实际示例的详细内容。更多信息请关注PHP中文网其他相关文章！
