ZH
日本語EspañolFrançaisPortuguêsTiếng ViệtEnglish简体中文한국어Deutsch
首页 / 架构与设计 / Kafka

Saga pattern 与 Kafka 分布式事务：预订旅游套餐示例

Cris The Dev · 2026/10/12 · 10 分钟阅读 · 98

1. 概述

当一个业务需要经过多个微服务，而每个服务都有独立的数据库时，你无法将所有操作封装在一个 @Transactional 中。Saga 模式解决了这个问题。与其用一个大事务，你可以运行一系列本地事务。每一步都有一个补偿行为（compensation），在后续步骤失败时用来撤销。

在本文中，你将使用 Spring Boot 3 和 Kafka 为旅游套餐预订业务实现一个 Saga orchestration 模式。读完本文后，你将拥有一个协调各步骤的 orchestrator、通过 Kafka 响应的 participant service，以及一套测试补偿流程。

2. 旅游套餐预订问题

一次旅游套餐预订包含三个步骤，每个步骤属于一个不同的服务：

  1. FlightService：预留机票座位。
  2. HotelService：预留酒店房间。
  3. PaymentService：扣款。

假设机票和房间已预留，但因为卡达到限额而支付失败。此时系统必须取消房间，然后取消机票。否则客户无需支付就占用了位置。

两阶段提交（2PC）可以解决这个问题，但它在整个事务期间锁定资源。它还要求所有服务支持 XA，而 Kafka 无法参与 XA。Saga 接受最终一致性：数据可能暂时不一致，但最终会保持一致。

3. Choreography 还是 Orchestration

Saga 有两种组织方式：

标准 Choreography Orchestration
协调方式 每个服务监听其他服务的事件 一个 orchestrator 发送命令，接收回复
业务流程 分散在多个服务 集中在一处
适用场景 2–3 个步骤，变化少 多个步骤，有复杂的补偿逻辑
风险 循环依赖，难以调试 Orchestrator 成为逻辑中枢

对于三个步骤和一个反向补偿流程，orchestration 更易读且易于测试。本文采用这种方式。Kafka topic 的组织如下：

tour.commands.flight   ← orchestrator 发送命令给 FlightService
tour.commands.hotel    ← orchestrator 发送命令给 HotelService
tour.commands.payment  ← orchestrator 发送命令给 PaymentService
tour.replies           ← 所有服务将结果发送给 orchestrator

4. 项目设置

4.1. 依赖

你需要添加 Spring for Apache Kafka。Spring Boot 为你管理版本：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

4.2. 配置

消息以 JSON 格式发送。配置序列化器和反序列化器如下：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: com.nonla.tour

JsonSerializer 将类名写入 header __TypeId__。在消费者端，JsonDeserializer 读取该 header 以了解需要创建什么对象。trusted.packages 限制了允许反序列化的包，以防止安全漏洞。

接下来，你声明各个 topic：

@Configuration
public class KafkaTopicConfig {

    @Bean
    KafkaAdmin.NewTopics sagaTopics() {
        return new KafkaAdmin.NewTopics(
            TopicBuilder.name("tour.commands.flight").partitions(3).build(),
            TopicBuilder.name("tour.commands.hotel").partitions(3).build(),
            TopicBuilder.name("tour.commands.payment").partitions(3).build(),
            TopicBuilder.name("tour.replies").partitions(3).build());
    }
}

5. 消息模型

你使用 Java 21 的 record 来表示命令和回复：

package com.nonla.tour;

public enum Step { FLIGHT, HOTEL, PAYMENT }

public enum Action { RESERVE, CANCEL }

public record SagaCommand(String bookingId, Step step, Action action, BigDecimal amount) {}

public record SagaReply(String bookingId, Step step, Action action,
                        boolean success, String reason) {}

每个消息携带 bookingId。这个值既是 saga 的标识符，也是发送到 Kafka 时的消息 key。这样，同一个预订的所有消息都会落入同一个分区并按正确的顺序处理。

6. 将 Saga 定义为状态机

最关键的部分是决定下一步的逻辑。你应该将这个逻辑与 Kafka 分离，以便用纯单元测试进行测试。

public enum SagaStatus { RUNNING, COMPENSATING, COMPLETED, CANCELLED }

public class SagaState {
    private final String bookingId;
    private final BigDecimal amount;
    private final Deque<Step> completed = new ArrayDeque<>();
    private SagaStatus status = SagaStatus.RUNNING;

    public SagaState(String bookingId, BigDecimal amount) {
        this.bookingId = bookingId;
        this.amount = amount;
    }

    public String bookingId() { return bookingId; }
    public BigDecimal amount() { return amount; }
    public Deque<Step> completed() { return completed; }
    public SagaStatus status() { return status; }
    public void status(SagaStatus status) { this.status = status; }
}

completed 是一个堆栈，保存已成功的步骤。当需要补偿时，你从堆栈顶部弹出，因此步骤会以相反的顺序被撤销。

public class TourBookingSaga {

    static final List<Step> STEPS = List.of(Step.FLIGHT, Step.HOTEL, Step.PAYMENT);

    public Optional<SagaCommand> start(SagaState s) {
        return Optional.of(command(s, STEPS.getFirst(), Action.RESERVE));
    }

    public Optional<SagaCommand> onReply(SagaState s, SagaReply r) {
        if (s.status() == SagaStatus.COMPLETED || s.status() == SagaStatus.CANCELLED) {
            return Optional.empty(); // saga 已结束，忽略重复的回复
        }
        if (r.action() == Action.RESERVE && r.success()) {
            s.completed().push(r.step());
            int idx = STEPS.indexOf(r.step());
            if (idx == STEPS.size() - 1) {
                s.status(SagaStatus.COMPLETED);
                return Optional.empty();
            }
            return Optional.of(command(s, STEPS.get(idx + 1), Action.RESERVE));
        }
        if (r.action() == Action.CANCEL && !r.success()) {
            return Optional.of(command(s, r.step(), Action.CANCEL)); // 重新尝试取消
        }
        s.status(SagaStatus.COMPENSATING);
        return compensateNext(s);
    }

    private Optional<SagaCommand> compensateNext(SagaState s) {
        if (s.completed().isEmpty()) {
            s.status(SagaStatus.CANCELLED);
            return Optional.empty();
        }
        return Optional.of(command(s, s.completed().pop(), Action.CANCEL));
    }

    private SagaCommand command(SagaState s, Step step, Action action) {
        return new SagaCommand(s.bookingId(), step, action, s.amount());
    }
}

onReply 函数处理四种情况：

  • RESERVE 成功：将该步骤记录到堆栈中。如果这是最后一步，saga 转换到 COMPLETED。否则，saga 向下一步发送 RESERVE 命令。
  • RESERVE 失败：saga 转换到 COMPENSATING 并开始取消最近成功的步骤。
  • CANCEL 成功：saga 取消前一步骤。当堆栈为空时，saga 转换到 CANCELLED。
  • CANCEL 失败：saga 重新发送取消命令。补偿不能中途放弃。

注意失败的步骤不在堆栈中。如果支付失败，没有什么需要退款，所以 saga 只需要取消房间和机票。

7. Orchestrator

Orchestrator 做两件事：向 Kafka 发送命令并从 tour.replies 监听回复。

@Service
public class TourBookingOrchestrator {

    private static final Logger log = LoggerFactory.getLogger(TourBookingOrchestrator.class);

    private final KafkaTemplate<String, Object> kafka;
    private final TourBookingSaga saga = new TourBookingSaga();
    private final Map<String, SagaState> store = new ConcurrentHashMap<>();

    public TourBookingOrchestrator(KafkaTemplate<String, Object> kafka) {
        this.kafka = kafka;
    }

    public void book(String bookingId, BigDecimal amount) {
        SagaState state = new SagaState(bookingId, amount);
        store.put(bookingId, state);
        saga.start(state).ifPresent(this::send);
    }

    @KafkaListener(topics = "tour.replies", groupId = "tour-orchestrator")
    public void onReply(SagaReply reply) {
        SagaState state = store.get(reply.bookingId());
        if (state == null) {
            return;
        }
        saga.onReply(state, reply).ifPresent(this::send);
        log.info("booking={} reply={}/{} ok={} -> status={}",
            reply.bookingId(), reply.step(), reply.action(), reply.success(), state.status());
    }

    private void send(SagaCommand cmd) {
        String topic = "tour.commands." + cmd.step().name().toLowerCase();
        kafka.send(topic, cmd.bookingId(), cmd);
    }

    public SagaStatus statusOf(String bookingId) {
        return store.get(bookingId).status();
    }
}

send() 使用 bookingId 作为 key。Topic 名称从 Step 推导而来，例如 HOTEL 变成 tour.commands.hotel。

注意：ConcurrentHashMap 仅用于演示。在生产环境中，saga 状态必须持久化到数据库中。如果 orchestrator 在中途重启，它需要读取状态以继续或补偿。

8. Participant Service

每个 participant 监听自己的 command topic，执行本地事务，然后发送回复。以 HotelService 为例：

@Component
public class HotelCommandHandler {

    private final KafkaTemplate<String, Object> kafka;
    private final Set<String> reservedRooms = ConcurrentHashMap.newKeySet();
    private final AtomicInteger availableRooms = new AtomicInteger(10);

    public HotelCommandHandler(KafkaTemplate<String, Object> kafka) {
        this.kafka = kafka;
    }

    @KafkaListener(topics = "tour.commands.hotel", groupId = "hotel-service")
    public void handle(SagaCommand cmd) {
        boolean ok = switch (cmd.action()) {
            case RESERVE -> reserve(cmd.bookingId());
            case CANCEL -> cancel(cmd.bookingId());
        };
        kafka.send("tour.replies", cmd.bookingId(),
            new SagaReply(cmd.bookingId(), Step.HOTEL, cmd.action(), ok, ok ? null : "没有可用房间"));
    }

    private boolean reserve(String bookingId) {
        if (reservedRooms.contains(bookingId)) {
            return true; // 已预留，消息被重新发送
        }
        if (availableRooms.getAndUpdate(n -> n > 0 ? n - 1 : n) == 0) {
            return false;
        }
        reservedRooms.add(bookingId);
        return true;
    }

    private boolean cancel(String bookingId) {
        if (reservedRooms.remove(bookingId)) {
            availableRooms.incrementAndGet();
        }
        return true; // 取消从未预留过的也算成功
    }
}

上述代码中的两个重要细节：

  • reserve() 是幂等的。Kafka 保证 at-least-once 交付，所以同一个命令可能到达两次。如果已为该 bookingId 预留了房间，handler 返回成功而不再扣减房间。
  • cancel() 也是幂等的，且始终成功。取消命令可能先于预留命令到达，或到达一个从未预留过房间的预订。这两种情况都不应该破坏数据。

FlightCommandHandler 和 PaymentCommandHandler 结构类似。对于 Payment，CANCEL 意味着退款（refund）。

当 PaymentService 拒绝支付时，orchestrator 的日志如下：

booking=BK-1001 reply=FLIGHT/RESERVE ok=true -> status=RUNNING
booking=BK-1001 reply=HOTEL/RESERVE ok=true -> status=RUNNING
booking=BK-1001 reply=PAYMENT/RESERVE ok=false -> status=COMPENSATING
booking=BK-1001 reply=HOTEL/CANCEL ok=true -> status=COMPENSATING
booking=BK-1001 reply=FLIGHT/CANCEL ok=true -> status=CANCELLED

房间先被取消，机票后被取消，这是与预留时相反的顺序。

9. Saga 流程测试

因为 TourBookingSaga 不依赖 Kafka，你可以使用 JUnit 5 和 AssertJ 进行测试，无需 broker：

class TourBookingSagaTest {

    private final TourBookingSaga saga = new TourBookingSaga();

    private SagaReply ok(Step step, Action action) {
        return new SagaReply("BK-1", step, action, true, null);
    }

    @Test
    void whenAllStepsSucceed_thenSagaCompleted() {
        SagaState s = new SagaState("BK-1", new BigDecimal("15000000"));

        assertThat(saga.start(s)).get().extracting(SagaCommand::step).isEqualTo(Step.FLIGHT);
        assertThat(saga.onReply(s, ok(Step.FLIGHT, Action.RESERVE)))
            .get().extracting(SagaCommand::step).isEqualTo(Step.HOTEL);
        assertThat(saga.onReply(s, ok(Step.HOTEL, Action.RESERVE)))
            .get().extracting(SagaCommand::step).isEqualTo(Step.PAYMENT);
        assertThat(saga.onReply(s, ok(Step.PAYMENT, Action.RESERVE))).isEmpty();

        assertThat(s.status()).isEqualTo(SagaStatus.COMPLETED);
    }

    @Test
    void whenPaymentFails_thenCompensateHotelThenFlight() {
        SagaState s = new SagaState("BK-1", new BigDecimal("15000000"));
        saga.start(s);
        saga.onReply(s, ok(Step.FLIGHT, Action.RESERVE));
        saga.onReply(s, ok(Step.HOTEL, Action.RESERVE));

        SagaCommand first = saga.onReply(s,
            new SagaReply("BK-1", Step.PAYMENT, Action.RESERVE, false, "卡被拒绝")).orElseThrow();
        assertThat(first).extracting(SagaCommand::step, SagaCommand::action)
            .containsExactly(Step.HOTEL, Action.CANCEL);
        assertThat(s.status()).isEqualTo(SagaStatus.COMPENSATING);

        SagaCommand second = saga.onReply(s, ok(Step.HOTEL, Action.CANCEL)).orElseThrow();
        assertThat(second).extracting(SagaCommand::step, SagaCommand::action)
            .containsExactly(Step.FLIGHT, Action.CANCEL);

        assertThat(saga.onReply(s, ok(Step.FLIGHT, Action.CANCEL))).isEmpty();
        assertThat(s.status()).isEqualTo(SagaStatus.CANCELLED);
    }
}

第二个测试检查了 saga 最重要的行为。支付失败应该导致取消酒店然后取消机票，且 saga 必须以 CANCELLED 状态结束。如果需要进行端到端测试，你可以使用 spring-kafka-test 中的 @EmbeddedKafka 或 Testcontainers。逻辑部分应该按上面的方式进行单元测试。

10. 常见错误

双重写入。 Participant 先写数据库，然后再发送回复。如果服务在这两个操作之间崩溃，saga 会卡住。标准的解决方案是 Transactional Outbox：在与业务数据相同的事务中将回复写入 outbox 表。然后一个单独的进程（或 Debezium CDC）将这些记录推送到 Kafka。

缺乏幂等性。 Consumer rebalance 或 retry 会重新发送消息。所有处理程序，包括补偿处理程序，都必须安全地处理重复。在 orchestrator 中，你应该额外保存待回复的步骤并忽略不匹配的回复。

不使用 key。 如果发送没有 key 的消息，同一预订的消息可能落入不同的分区。这样取消命令可能先于预留命令被处理。

缺乏隔离性。 Saga 没有 ACID 中的隔离性。当 saga 运行时，其他服务可能看到房间“已预留”，尽管稍后会被取消。你应该使用中间状态如 PENDING（语义锁）并仅在 saga 完成时转换为 CONFIRMED。

缺少超时机制。 如果一个服务永远不回复，saga 会一直等待。你需要一个定期扫描超期 saga 进行补偿的任务。对于持续失败的 CANCEL 命令，应该限制重试次数并将消息转移到死信队列供人工处理。

**注意：**如果所有数据都在同一个数据库中，不要使用 Saga。普通的 @Transactional 更简单、更安全。Saga 也不适合那些不接受中间状态的业务，或有无法补偿的步骤。例如，已发送的确认邮件无法“撤回”。你应该将这类步骤放在 saga 的最后。

11. 总结

在本文中，你为旅游套餐预订业务实现了 Kafka 的 Saga orchestration 模式：

  • Saga 逻辑被分离成一个纯粹的状态机，可以在不需要 broker 的情况下进行测试。
  • Orchestrator 通过 Kafka 发送命令并接收回复，使用 bookingId 作为 key 来维护顺序。
  • 补偿步骤通过堆栈按相反的顺序执行。
  • Participant 以幂等方式处理命令。

将 saga 投入生产时，你需要补充三个方面：将 saga 状态持久化到数据库，使用 Transactional Outbox 来避免双重写入，以及拥有超时机制和死信队列。这些是决定你的 saga 是否真正可靠的关键部分。

kafkasagamicroservicesspring-bootdistributed-transaction
作者的应用HoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

同主题文章