【系统设计】高效的分布式系统：使用 Spring Boot 和 Kafka 实现 Saga 模式

🕗 发布于 2024-11-05 10:59 spring boot kafka linq 架构云原生

在现代分布式系统中，管理跨多个服务的长事务至关重要。传统的分布式事务解决方案往往面临性能瓶颈和复杂性问题，而 Saga 模式 作为一种灵活高效的解决方案，逐渐受到开发者的青睐。本文将探讨如何利用 Spring Boot 和 Kafka 实现 Saga 模式，并详细介绍事务补偿机制，帮助你构建稳定可靠的分布式系统。

什么是 Saga 模式？

原理介绍

在微服务架构中，一个业务流程通常涉及多个独立的服务。这些服务必须协同工作以完成完整的业务操作。例如，用户下单可能需要订单服务、支付服务和库存服务的合作。然而，跨服务操作通常涉及复杂的事务管理，传统的分布式事务（如两阶段提交）不仅效率低下，还难以扩展和维护。

Saga 模式 提供了一种替代方案，通过将一个长事务分解为一系列的本地事务，并通过事件或命令进行协调，从而实现最终一致性。这种方法不仅提高了系统的可扩展性，还简化了事务管理。

解决的问题及其重要性

Saga 模式解决了以下问题：

分布式事务管理：通过拆分事务，避免了传统分布式事务的性能和复杂性问题。
系统可扩展性：各服务独立运行，易于扩展和维护。
错误恢复：通过补偿机制，确保在步骤失败时，系统能恢复到一致状态。

在现代微服务架构中，确保跨服务操作的可靠性和一致性至关重要。Saga 模式提供了一个高效且灵活的解决方案，使系统在面对复杂业务流程和潜在错误时能够稳定运行。

Saga 模式的组成部分与实现方法

Saga 模式主要有两种实现方式：Choreography（编排） 和 Orchestration（指挥）。下面将详细介绍这两种模式，并展示如何使用 Spring Boot 和 Kafka 实现它们，包括事务补偿机制。

架构图

编排模式

+----------------+        +----------------+        +----------------+
|  OrderService  |        | PaymentService |        | InventoryService|
+----------------+        +----------------+        +----------------+
        |                          |                         |
        |  CreateOrderCommand      |                         |
        |------------------------->|                         |
        |                          |                         |
        |      OrderCreatedEvent   |                         |
        |<-------------------------|                         |
        |                          |                         |
        |                          |      PaymentCommand     |
        |                          |------------------------>|
        |                          |                         |
        |                          |    PaymentProcessedEvent |
        |                          |<------------------------|
        |                          |                         |
        |      InventoryCommand    |                         |
        |------------------------->|                         |
        |                          |                         |
        |                          |    InventoryUpdatedEvent|
        |<-------------------------|                         |
        |                          |                         |

指挥模式

+----------------+          +----------------+          +----------------+
| SagaOrchestrator|          | OrderService  |          | PaymentService |
+----------------+          +----------------+          +----------------+
        |                        |                           |
        |   CreateOrderCommand   |                           |
        |----------------------->|                           |
        |                        |                           |
        |   OrderCreatedEvent    |                           |
        |<-----------------------|                           |
        |                        |                           |
        |   PaymentCommand       |                           |
        |--------------------------------------------------->|
        |                        |                           |
        |   PaymentApprovedEvent |                           |
        |<---------------------------------------------------|
        |                        |                           |
        |                        |                           |
        |   InventoryCommand     |                           |
        |----------------------->|                           |
        |                        |                           |

1. 编排模式

在 编排模式 下，各服务通过事件进行通信和协调，没有中央控制器。每个服务独立地监听和发布事件，以完成整个业务流程。

1.1 组成部分

事件定义：服务之间传递的消息，如 OrderCreatedEvent、PaymentProcessedEvent 等。
Kafka 生产者与消费者：用于事件的发布和订阅。
各服务逻辑：根据收到的事件执行相应的操作，并发布下一个事件。

1.2 实现步骤与代码

1.2.1 定义事件

// OrderCreatedEvent.java
public class OrderCreatedEvent {
    private String orderId;
    private String product;
    private int quantity;
    // getters and setters
}

// PaymentProcessedEvent.java
public class PaymentProcessedEvent {
    private String orderId;
    private boolean success;
    // getters and setters
}

// PaymentFailedEvent.java
public class PaymentFailedEvent {
    private String orderId;
    private String reason;
    // getters and setters
}

// OrderCancelledEvent.java
public class OrderCancelledEvent {
    private String orderId;
    // getters and setters
}

1.2.2 配置 Kafka

# application.yml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: saga-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

1.2.3 实现 OrderService

@Service
public class OrderService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private ObjectMapper mapper = new ObjectMapper();

    @KafkaListener(topics = "create-order-command", groupId = "order-service-group")
    public void handleCreateOrder(String message) throws JsonProcessingException {
        CreateOrderCommand command = mapper.readValue(message, CreateOrderCommand.class);
        
        // 创建订单逻辑
        // TODO: 保存订单到数据库

        // 发布 OrderCreatedEvent
        OrderCreatedEvent event = new OrderCreatedEvent();
        event.setOrderId(command.getOrderId());
        event.setProduct(command.getProduct());
        event.setQuantity(command.getQuantity());
        String eventMsg = mapper.writeValueAsString(event);
        kafkaTemplate.send("order-created", eventMsg);
    }

    @KafkaListener(topics = "payment-failed", groupId = "saga-group")
    public void handlePaymentFailed(String message) throws JsonProcessingException {
        PaymentFailedEvent failedEvent = mapper.readValue(message, PaymentFailedEvent.class);
        
        // 取消订单逻辑
        cancelOrder(failedEvent.getOrderId());
        
        // 发布 OrderCancelledEvent
        OrderCancelledEvent cancelledEvent = new OrderCancelledEvent();
        cancelledEvent.setOrderId(failedEvent.getOrderId());
        String cancelledMsg = mapper.writeValueAsString(cancelledEvent);
        kafkaTemplate.send("order-cancelled", cancelledMsg);
    }

    private void cancelOrder(String orderId) {
        // TODO: 取消订单逻辑
    }
}

1.2.4 实现 PaymentService

@Service
public class PaymentService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private ObjectMapper mapper = new ObjectMapper();

    @KafkaListener(topics = "order-created", groupId = "saga-group")
    public void handleOrderCreated(String message) throws JsonProcessingException {
        OrderCreatedEvent event = mapper.readValue(message, OrderCreatedEvent.class);
        
        // 处理支付逻辑
        boolean success = processPayment(event.getOrderId(), event.getQuantity());

        if (success) {
            // 发布 PaymentProcessedEvent
            PaymentProcessedEvent paymentEvent = new PaymentProcessedEvent(event.getOrderId(), true);
            String paymentMsg = mapper.writeValueAsString(paymentEvent);
            kafkaTemplate.send("payment-processed", paymentMsg);
        } else {
            // 发布 PaymentFailedEvent
            PaymentFailedEvent failedEvent = new PaymentFailedEvent();
            failedEvent.setOrderId(event.getOrderId());
            failedEvent.setReason("Payment processing failed.");
            String failedMsg = mapper.writeValueAsString(failedEvent);
            kafkaTemplate.send("payment-failed", failedMsg);
        }
    }

    private boolean processPayment(String orderId, int quantity) {
        // TODO: 实现支付逻辑，模拟支付失败
        return false;
    }
}

1.2.5 实现 InventoryService

@Service
public class InventoryService {
    @KafkaListener(topics = "order-cancelled", groupId = "saga-group")
    public void handleOrderCancelled(String message) throws JsonProcessingException {
        OrderCancelledEvent cancelledEvent = new ObjectMapper().readValue(message, OrderCancelledEvent.class);
        
        // 回滚库存逻辑
        rollbackInventory(cancelledEvent.getOrderId());
    }

    private void rollbackInventory(String orderId) {
        // TODO: 实现库存回滚逻辑
    }
}

1.3 事务补偿机制

在 编排模式 中，当某个服务的操作失败时，需要通过发布补偿事件来反向撤销之前的操作。上述代码中，PaymentService 在支付失败时发布 PaymentFailedEvent，OrderService 监听该事件并执行订单取消逻辑，随后发布 OrderCancelledEvent，最后 InventoryService 监听并回滚库存。

2. 指挥模式

在 指挥模式 下，存在一个中央的 Saga 管理器（Orchestrator），负责调度和协调各个服务的操作，并在需要时触发补偿机制。

2.1 组成部分

Saga Orchestrator：负责整个事务流程的控制和协调。
命令和事件定义：如 CreateOrderCommand、PaymentCommand 等。
Kafka 生产者与消费者：用于命令和事件的发布与订阅。
各服务逻辑：根据接收到的命令执行操作，并发布相应的事件。

2.2 实现步骤与代码

2.2.1 定义命令和事件

// CreateOrderCommand.java
public class CreateOrderCommand {
    private String orderId;
    private String product;
    private int quantity;
    // getters and setters
}

// OrderCreatedEvent.java
public class OrderCreatedEvent {
    private String orderId;
    // getters and setters
}

// PaymentCommand.java
public class PaymentCommand {
    private String orderId;
    private double amount;
    // getters and setters
}

// PaymentApprovedEvent.java
public class PaymentApprovedEvent {
    private String orderId;
    // getters and setters
}

// PaymentRejectedEvent.java
public class PaymentRejectedEvent {
    private String orderId;
    private String reason;
    // getters and setters
}

// CancelOrderCommand.java
public class CancelOrderCommand {
    private String orderId;
    // getters and setters
}

// OrderCancelledEvent.java
public class OrderCancelledEvent {
    private String orderId;
    // getters and setters
}

2.2.2 实现 SagaOrchestrator

@Service
public class SagaOrchestrator {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    private ObjectMapper mapper = new ObjectMapper();

    @KafkaListener(topics = "order-created", groupId = "saga-orchestrator-group")
    public void handleOrderCreated(String message) throws JsonProcessingException {
        OrderCreatedEvent event = mapper.readValue(message, OrderCreatedEvent.class);
        
        try {
            // 发送 PaymentCommand
            PaymentCommand paymentCommand = new PaymentCommand();
            paymentCommand.setOrderId(event.getOrderId());
            paymentCommand.setAmount(calculateAmount(event.getOrderId()));
            String paymentCmd = mapper.writeValueAsString(paymentCommand);
            kafkaTemplate.send("payment-command", paymentCmd);
        } catch (Exception e) {
            // 发送补偿命令
            sendCancelOrderCommand(event.getOrderId());
        }
    }

    @KafkaListener(topics = "payment-approved", groupId = "saga-orchestrator-group")
    public void handlePaymentApproved(String message) throws JsonProcessingException {
        PaymentApprovedEvent event = mapper.readValue(message, PaymentApprovedEvent.class);
        
        // 继续后续操作，如库存更新
        // TODO: 发送其他命令或处理逻辑
    }

    @KafkaListener(topics = "payment-rejected", groupId = "saga-orchestrator-group")
    public void handlePaymentRejected(String message) throws JsonProcessingException {
        PaymentRejectedEvent event = mapper.readValue(message, PaymentRejectedEvent.class);
        
        // 发送补偿命令
        sendCancelOrderCommand(event.getOrderId());
    }

    public void startSaga(Order order) throws JsonProcessingException {
        // 发送 CreateOrderCommand
        CreateOrderCommand createCommand = new CreateOrderCommand();
        createCommand.setOrderId(order.getId());
        createCommand.setProduct(order.getProduct());
        createCommand.setQuantity(order.getQuantity());
        String createCmd = mapper.writeValueAsString(createCommand);
        kafkaTemplate.send("create-order-command", createCmd);
    }

    private void sendCancelOrderCommand(String orderId) throws JsonProcessingException {
        CancelOrderCommand cancelCommand = new CancelOrderCommand();
        cancelCommand.setOrderId(orderId);
        String cancelMsg = mapper.writeValueAsString(cancelCommand);
        kafkaTemplate.send("cancel-order-command", cancelMsg);
    }

    private double calculateAmount(String orderId) {
        // TODO: 实现金额计算逻辑
        return 100.0;
    }
}

2.2.3 修改 OrderService

@Service
public class OrderService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    private ObjectMapper mapper = new ObjectMapper();

    @KafkaListener(topics = "create-order-command", groupId = "order-service-group")
    public void handleCreateOrder(String message) throws JsonProcessingException {
        CreateOrderCommand command = mapper.readValue(message, CreateOrderCommand.class);
        
        // 创建订单逻辑
        // TODO: 保存订单到数据库

        // 发布 OrderCreatedEvent
        OrderCreatedEvent event = new OrderCreatedEvent();
        event.setOrderId(command.getOrderId());
        String eventMsg = mapper.writeValueAsString(event);
        kafkaTemplate.send("order-created", eventMsg);
    }

    @KafkaListener(topics = "cancel-order-command", groupId = "order-service-group")
    public void handleCancelOrder(String message) throws JsonProcessingException {
        CancelOrderCommand command = mapper.readValue(message, CancelOrderCommand.class);
        
        // 取消订单逻辑
        cancelOrder(command.getOrderId());
        
        // 发布 OrderCancelledEvent
        OrderCancelledEvent event = new OrderCancelledEvent();
        event.setOrderId(command.getOrderId());
        String cancelledMsg = mapper.writeValueAsString(event);
        kafkaTemplate.send("order-cancelled", cancelledMsg);
    }

    private void cancelOrder(String orderId) {
        // TODO: 实现取消订单逻辑
    }
}

2.2.4 修改 PaymentService

@Service
public class PaymentService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    private ObjectMapper mapper = new ObjectMapper();

    @KafkaListener(topics = "payment-command", groupId = "payment-service-group")
    public void handlePaymentCommand(String message) throws JsonProcessingException {
        PaymentCommand command = mapper.readValue(message, PaymentCommand.class);
        
        // 处理支付逻辑
        boolean approved = processPayment(command.getOrderId(), command.getAmount());

        if (approved) {
            // 发布 PaymentApprovedEvent
            PaymentApprovedEvent event = new PaymentApprovedEvent();
            event.setOrderId(command.getOrderId());
            String approvedMsg = mapper.writeValueAsString(event);
            kafkaTemplate.send("payment-approved", approvedMsg);
        } else {
            // 发布 PaymentRejectedEvent
            PaymentRejectedEvent rejectedEvent = new PaymentRejectedEvent();
            rejectedEvent.setOrderId(command.getOrderId());
            rejectedEvent.setReason("Payment was rejected.");
            String rejectedMsg = mapper.writeValueAsString(rejectedEvent);
            kafkaTemplate.send("payment-rejected", rejectedMsg);
        }
    }

    private boolean processPayment(String orderId, double amount) {
        // TODO: 实现支付逻辑，模拟支付失败
        return false;
    }
}

2.2.5 实现 InventoryService

@Service
public class InventoryService {
    @KafkaListener(topics = "order-cancelled", groupId = "saga-orchestrator-group")
    public void handleOrderCancelled(String message) throws JsonProcessingException {
        OrderCancelledEvent cancelledEvent = new ObjectMapper().readValue(message, OrderCancelledEvent.class);
        
        // 回滚库存逻辑
        rollbackInventory(cancelledEvent.getOrderId());
    }

    private void rollbackInventory(String orderId) {
        // TODO: 实现库存回滚逻辑
    }
}

2.3 事务补偿机制

在 指挥模式 中，Saga Orchestrator 作为中央控制器，负责监控事务流程并在发生错误时触发补偿操作。例如，当 PaymentService 处理支付失败时，发布 PaymentRejectedEvent，Saga Orchestrator 监听到该事件后，发送 CancelOrderCommand 给 OrderService 执行订单取消操作，确保系统一致性。

总结

Saga 模式为分布式系统中的长事务管理提供了一种高效且灵活的解决方案。通过编排与指挥两种实现方式，开发者可以根据具体业务需求和系统架构选择最合适的方式。

编排模式 适用于服务间关系较为松散、需要高扩展性的场景，各服务通过事件进行独立协调，但补偿逻辑较为分散。
指挥模式 适用于需要集中控制事务流程、易于追踪和调试的场景，Saga Orchestrator 作为中央管理者，补偿逻辑统一管理，但可能成为系统的瓶颈。

无论选择哪种模式，事务补偿机制 都是确保系统可靠性和一致性的关键。通过合理设计补偿逻辑，可以有效应对分布式环境下的各种故障和异常，提升系统的健壮性。

借助 Spring Boot 和 Kafka 强大的生态和工具支持，实现 Saga 模式变得更加便捷和高效。希望本文能够帮助你深入理解 Saga 模式，并在实际项目中灵活应用，打造稳定可靠的分布式系统。

参考资料

版权声明

本文为原创内容，转载请注明出处。

原文地址：https://blog.csdn.net/yhkal/article/details/143373469

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：20221010-Behere-VR审稿意见
下一篇：创新材料科技：铜冷却壁助力高炉节能降耗

【工具类】JAVA开发从SFTP服务器下载文件
SFTP工具类，打成JAR文件后直接放到服务器上执行。
阅读更多2024-11-05
【linux】HTTPS 协议原理
HHTPS协议
阅读更多2024-11-05
鸿蒙HarmonyOS开发：给应用添加基础类型通知和进度条类型通知（API 12）
通知旨在让用户以合适的方式及时获得有用的新消息，帮助用户高效地处理任务。显示接收到的短消息、即时消息等。显示应用的推送消息，如广告、版本更新等。显示当前正在进行的事件，如下载等。
阅读更多2024-11-05
Centos 网络接口打vlan标签
本次使用给bond打vlan标签，其实其他普通接口也一样。查询802.1q 模块是否加载。2.1. 创建bond配置文件。本次实验使用bond1 模式。给bond0 接口打标签。重启网络查看vlan标
阅读更多2024-11-05
libaom 源码分析：帧内方向预测模式
libaom 源码分析：帧内方向预测模式
阅读更多2024-11-05
java版CRM客户关系管理系统源码-CRM客户关系管理系统的技术架构与功能实现
本CRM客户关系管理系统通过整合待办事项、线索、客户、联系人、商机、合同、回款等多个功能模块，为企业提供了一套全面、高效的客户关系管理解决方案。系统采用先进的技术架构和灵活的配置方式，能够满足企业
阅读更多2024-11-05
【数学二】线性代数-矩阵-初等变换、初等矩阵
【数学二】线性代数-矩阵-初等变换、初等矩阵
阅读更多2024-11-05
京准时钟：无人机卫星信号安全防护隔离装置
京准时钟：无人机卫星信号安全防护隔离装置
阅读更多2024-11-05
Blender进阶：着色器节点
着色点着色点UV坐标取得UV中的x值，此值介于【0,1】将X值转成一个颜色颜色作为基础色，完成着色。每一个着色点，都执行一遍这套流程。最终完成整个面的着色。在着色时，每个点的颜色根据UV坐标计算得到
阅读更多2024-11-05
WPF开发，将引用的dll在编译时指向单独的文件夹。
在这种情况下，它将获取"引用的程序集"解决方案文件夹所引用的新dll，并在提交它们之前自动添加源代码控制跟踪(即，如果像我一样从Visual Studio中的解决方案资源管理器中提
阅读更多2024-11-05