Debezium日常分享系列之：使用 Outbox 模式实现可靠的微服务数据交换

作为其业务逻辑的一部分，微服务通常不仅需要更新自己的本地数据存储，还需要通知其他服务发生的数据更改。离线箱模式描述了一种让服务以安全和一致的方式执行这两个任务的方法；它为源服务提供了即时的“读取自己的写入”语义，同时提供可靠、最终一致的服务边界之间的数据交换。

更新（2019年9月13日）：为了简化离线箱模式的使用，Debezium现在提供了一个即用型的SMT用于路由离线箱事件。本文讨论的自定义SMT不再需要。

如果你构建过一些微服务，你可能会同意关于它们最困难的部分就是数据：微服务不是孤立存在的，它们往往需要在彼此之间传播数据和数据更改。

例如，考虑一个管理采购订单的微服务：当一个新订单被下达时，关于该订单的信息可能需要传递给发货服务（以便它可以组装一个或多个订单的货物）和客户服务（以便它可以基于新订单更新客户的总信用余额）。

有不同的方法可以让订单服务知道其他两个服务的新采购订单；例如，它可以调用这些服务提供的某些REST、grpc或其他（同步）API。然而，这可能会导致一些不希望的耦合：发送服务必须知道要调用哪些其他服务以及在哪里找到它们。它还必须准备好这些服务暂时不可用的情况。在这里，服务网格（如Istio）可以提供帮助，通过提供请求路由、重试、断路器等功能。

任何同步方法的一般问题是，一个服务在没有调用其他服务的情况下无法正常工作。虽然缓冲和重试可能有助于在其他服务只需要被通知某些事件的情况下，但如果一个服务实际上需要查询其他服务的信息，这种情况就不适用。例如，当下达一个采购订单时，订单服务可能需要从库存服务中获取购买商品库存的次数信息。

这种同步方法的另一个缺点是它缺乏可重放性，即新的消费者可以在事件发送后到达并仍然能够从头开始消费整个事件流。

这两个问题可以通过使用异步数据交换方法来解决：即通过一个持久的消息日志（如Apache Kafka）来传播订单、库存和其他服务的事件。通过订阅这些事件流，每个服务都会收到关于其他服务数据更改的通知。它可以对这些事件做出反应，并在需要时在自己的数据存储中创建该数据的本地表示，使用适合自己需求的表示。例如，这样的视图可能是去规范化的，以有效地支持特定的访问模式，或者它可能只包含原始数据的子集，这对于消费服务是相关的。

持久的日志还支持可重放性，即可以根据需要添加新的消费者，以支持您最初可能没有考虑到的用例，而无需触及源服务。例如，考虑一个数据仓库，应该保存有关所有已下达的订单的信息，或者基于Elasticsearch的采购订单的全文搜索功能。一旦采购订单事件在Kafka的主题中（Kafka的主题保留策略设置可用于确保事件在主题中保留的时间足够用于给定的用例和业务需求），新的消费者可以订阅、处理主题并在微服务的数据库、搜索索引、数据仓库等中实现所有数据的视图。

注意：处理Topic增长，根据数据量（记录数量和大小，变更频率），保留事件在话题中长期甚至无限期可能会不可行。很多时候，与给定数据项（例如特定采购订单）相关的一些甚至所有事件，在一定时间点之后从业务角度来看可能会被删除。

双写的问题

为了提供其功能，微服务通常会有自己的本地数据存储。例如，订单服务可能使用关系数据库来持久化有关采购订单的信息。当下订单时，这可能会导致在服务的数据库中的PurchaseOrder表中进行INSERT操作。同时，服务可能希望将有关新订单的事件发送到Apache Kafka，以便将该信息传播给其他感兴趣的服务。

然而，仅仅发出这两个请求可能会导致潜在的不一致性。原因在于我们无法拥有一个跨服务数据库和Apache Kafka的共享事务，因为后者不支持加入分布式（XA）事务。因此，在不幸的情况下，我们可能会发现新的采购订单已在本地数据库中持久化，但未将相应的消息发送到Kafka（例如由于某些网络问题）。或者，反过来，我们可能已将消息发送到Kafka，但未能将采购订单持久化到本地数据库中。这两种情况都是不可取的；这可能导致对于一个看似成功下单的订单没有创建发货记录。或者创建了一个发货记录，但订单服务本身没有关于相应采购订单的迹象。

那么如何避免这种情况呢？答案是仅修改两个资源中的一个（数据库或Apache Kafka），并以最终一致的方式驱动第二个资源的更新。让我们首先考虑只写入Apache Kafka的情况。

当接收到新的采购订单时，订单服务不会同步将其插入数据库；相反，它只会将描述新订单的事件发送到Kafka主题。因此，一次只修改一个资源，如果出现问题，我们将立即发现并向订单服务的调用方报告请求失败。
同时，服务本身将订阅该Kafka主题。这样，当主题中有新的消息到达时，它将收到通知，并可以将新的采购订单持久化到其数据库中。然而，在这里存在一个微妙的挑战，即缺乏“读自己写”的语义。例如，假设订单服务还具有根据给定客户搜索所有采购订单的API。在下订单后立即调用该API时，由于从Kafka主题处理消息的异步性质，可能会发生采购订单尚未在服务的数据库中持久化，因此不会被该查询返回。这可能导致非常混乱的用户体验，因为用户可能会在其购物历史中错过新下的订单。有办法解决这个问题，例如，服务可以将新下的采购订单保存在内存中，并基于此回答后续查询。然而，当实现更复杂的查询或考虑到订单服务可能还包含在集群设置中的多个节点时，这很快就会变得复杂起来，这将要求在集群内传播该数据。

那么，当仅同步写入数据库并基于此驱动向Apache Kafka导出消息时，情况会怎样呢？这就是出箱模式的用武之地。

出箱模式

这种方法的想法是在服务的数据库中有一个“出箱”表。当接收到下订单请求时，不仅会在PurchaseOrder表中进行插入操作，而且在同一事务中，还会将表示要发送的事件的记录插入到该出箱表中。

该记录描述了服务中发生的事件，例如，它可以是表示已下新订单的JSON结构，包括订单本身的数据、订单行以及上下文信息，如用例标识符。通过显式地通过出箱表中的记录发出事件，可以确保事件以适合外部消费者的方式结构化。这也有助于确保事件消费者在修改内部领域模型或PurchaseOrder表时不会出现问题。

异步进程监视该表以获取新条目。如果有任何条目，它将将事件作为消息传播到Apache Kafka。这给我们带来了非常好的特性平衡：通过同步写入PurchaseOrder表，源服务可以受益于“读自己写”的语义。对采购订单的后续查询将在第一次事务提交后返回新持久化的订单。同时，我们通过Apache Kafka获得可靠、异步、最终一致的数据传播到其他服务。

现在，出箱模式实际上不是一个新的想法。实际上，即使在使用JMS风格的消息代理时，它实际上可以参与分布式事务，但出箱模式也可以成为避免与远程资源（如消息代理）的停机耦合和潜在影响的首选选项。您还可以在Chris Richardson的优秀网站microservices.io上找到有关该模式的描述。

然而，该模式没有得到应有的关注，特别是在微服务的背景下非常有用。正如我们将看到的，可以使用变更数据捕获和Debezium以非常优雅和高效的方式实现出箱模式。接下来，让我们探讨如何实现。

基于变更数据捕获的实现

基于日志的变更数据捕获（Change Data Capture，CDC）非常适合捕获出箱表中的新条目并将其流式传输到Apache Kafka。与任何基于轮询的方法相比，事件捕获在几乎实时的情况下开销非常低。Debezium提供了用于多个数据库（如MySQL、Postgres和SQL Server）的CDC连接器。以下示例将使用Postgres的Debezium连接器。

您可以在GitHub上找到完整的示例源代码。有关构建和运行示例代码的详细信息，请参阅README.md。该示例围绕两个微服务order-service和shipment-service展开。两者都使用Java实现，使用CDI作为组件模型，并使用JPA/Hibernate访问各自的数据库。订单服务在WildFly上运行，并公开一个简单的REST API，用于下订单和取消特定订单行。它使用Postgres数据库作为其本地数据存储。发货服务基于Thorntail；通过Apache Kafka，它接收订单服务导出的事件，并在自己的MySQL数据库中创建相应的发货记录。为了捕获出箱表中的任何新事件并将其传播到Apache Kafka，Debezium会追踪订单服务的Postgres数据库的事务日志（“预写日志”，WAL）。

解决方案的整体架构如下图所示：

请注意，该模式与这些特定的实现选择没有任何关系。它同样可以使用替代技术来实现，例如 Spring Boot（例如利用 Spring Data 对域事件的支持）、普通 JDBC 或 Java 以外的其他编程语言。

现在让我们仔细看看该解决方案的一些相关组件。

发件箱表

发件箱表位于订单服务的数据库中，其结构如下：

Column        |          Type          | Modifiers
--------------+------------------------+-----------
id            | uuid                   | not null
aggregatetype | character varying(255) | not null
aggregateid   | character varying(255) | not null
type          | character varying(255) | not null
payload       | jsonb                  | not null

它的列包括：

• id：每条消息的唯一标识，可以用于消费者在故障后重新读取消息时检测任何重复事件。在创建新事件时生成。
• aggregatetype：与给定事件相关的聚合根的类型；根据领域驱动设计的概念，导出的事件应该引用一个聚合根（aggregate），其中聚合根提供了访问聚合中任何实体的唯一入口点。例如，可以是“采购订单”或“客户”。此值将用于将事件路由到Kafka中相应的主题，因此与采购订单相关的所有事件都将有一个主题，与客户相关的所有事件都将有一个主题，以此类推。注意，与包含在这样的聚合中的子实体有关的事件也应使用相同的类型。例如，表示取消单个订单行的事件（它是采购订单聚合的一部分）也应使用其聚合根的类型“order”，以确保此事件也将进入“order” Kafka主题。
• aggregateid：受给定事件影响的聚合根的id；例如，可以是一个采购订单的id或客户id；与聚合类型类似，与聚合中的子实体相关的事件应使用包含聚合根的id，例如，订单行取消事件应使用采购订单id。此id将在后续用作Kafka消息的键。这样，与一个聚合根或其包含的任何子实体相关的所有事件都将进入该Kafka主题的同一个分区，从而确保该主题的消费者将按照它们产生的顺序消费所有与同一个聚合相关的事件。
• type：事件的类型，例如“创建订单”或“取消订单行”。允许消费者触发适当的事件处理程序。
• payload：包含实际事件内容的JSON结构，例如包含一个采购订单、购买者信息、包含的订单行、它们的价格等。

发送事件到信箱

为了将事件“发送”到信箱，订单服务中的代码通常可以只需向信箱表中插入一条记录。然而，如果需要，最好选择一个稍微抽象一些的API，以便更轻松地调整信箱的实现细节。CDI事件对此非常有用。它们可以在应用代码中触发，并且将由信箱事件发送器同步处理，该发送器将执行所需的插入操作到信箱表中。

所有信箱事件类型都应实现以下契约，类似于之前显示的信箱表的结构：

public interface ExportedEvent {

    String getAggregateId();
    String getAggregateType();
    JsonNode getPayload();
    String getType();
}

为了产生这样的事件，应用程序代码使用注入的事件实例，例如这里的 OrderService 类：

@ApplicationScoped
public class OrderService {

    @PersistenceContext
    private EntityManager entityManager;

    @Inject
    private Event<ExportedEvent> event;

    @Transactional
    public PurchaseOrder addOrder(PurchaseOrder order) {
        order = entityManager.merge(order);

        event.fire(OrderCreatedEvent.of(order));
        event.fire(InvoiceCreatedEvent.of(order));

        return order;
    }

    @Transactional
    public PurchaseOrder updateOrderLine(long orderId, long orderLineId,
            OrderLineStatus newStatus) {
        // ...
    }
}

在addOrder()方法中，使用JPA实体管理器将传入的订单持久化到数据库，并使用注入的事件触发相应的OrderCreatedEvent和InvoiceCreatedEvent。再次强调，尽管有“事件”的概念，但这两个操作发生在同一个事务中。也就是说，在此事务中，将向数据库插入三条记录：一条记录在采购订单表中，两条记录在信箱表中。

实际的事件实现非常直接；以下是OrderCreatedEvent类的示例：

public class OrderCreatedEvent implements ExportedEvent {

    private static ObjectMapper mapper = new ObjectMapper();

    private final long id;
    private final JsonNode order;

    private OrderCreatedEvent(long id, JsonNode order) {
        this.id = id;
        this.order = order;
    }

    public static OrderCreatedEvent of(PurchaseOrder order) {
        ObjectNode asJson = mapper.createObjectNode()
                .put("id", order.getId())
                .put("customerId", order.getCustomerId())
                .put("orderDate", order.getOrderDate().toString());

        ArrayNode items = asJson.putArray("lineItems");

        for (OrderLine orderLine : order.getLineItems()) {
        items.add(
                mapper.createObjectNode()
                .put("id", orderLine.getId())
                .put("item", orderLine.getItem())
                .put("quantity", orderLine.getQuantity())
                .put("totalPrice", orderLine.getTotalPrice())
                .put("status", orderLine.getStatus().name())
            );
        }

        return new OrderCreatedEvent(order.getId(), asJson);
    }

    @Override
    public String getAggregateId() {
        return String.valueOf(id);
    }

    @Override
    public String getAggregateType() {
        return "Order";
    }

    @Override
    public String getType() {
        return "OrderCreated";
    }

    @Override
    public JsonNode getPayload() {
        return order;
    }
}

请注意 Jackson 的 ObjectMapper 是如何用于创建事件负载的 JSON 表示的。

现在让我们看一下使用任何触发的 ExportedEvent 并对发件箱表进行相应写入的代码：

@ApplicationScoped
public class EventSender {

    @PersistenceContext
    private EntityManager entityManager;

    public void onExportedEvent(@Observes ExportedEvent event) {
        OutboxEvent outboxEvent = new OutboxEvent(
                event.getAggregateType(),
                event.getAggregateId(),
                event.getType(),
                event.getPayload()
        );

        entityManager.persist(outboxEvent);
        entityManager.remove(outboxEvent);
    }
}

这非常简单：对于每个事件，CDI运行时将调用onExportedEvent()方法。OutboxEvent实体的一个实例被持久化到数据库中 - 然后立即删除！

这可能会令人惊讶。但是如果记得日志基于CDC的工作方式，这就有意义了：它不会检查数据库中表的实际内容，而是追踪追加式事务日志。一旦事务提交，对persist()和remove()的调用将在日志中创建一个INSERT和一个DELETE条目。此后，Debezium将处理这些事件：对于任何INSERT，将发送一个带有事件负载的消息到Apache Kafka。另一方面，可以忽略DELETE事件，因为从信箱表中删除只是一种纯技术性操作，不需要将其传播到消息代理。因此，我们能够通过CDC捕获添加到信箱表中的事件，但在查看表本身的内容时，它始终为空。这意味着不需要为该表增加额外的磁盘空间（除了日志文件元素，它们将在某个时候自动丢弃），也不需要单独的维护进程来防止其无限增长。

注册Debezium连接器

在实现信箱功能之后，现在是时候注册Debezium Postgres连接器了，以便它可以捕获信箱表中的任何新事件并将它们传递到Apache Kafka。可以通过向Kafka Connect的REST API发送以下JSON请求来完成这个操作：

{
    "name": "outbox-connector",
    "config": {
        "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
        "tasks.max" : "1",
        "database.hostname" : "order-db",
        "database.port" : "5432",
        "database.user" : "postgresuser",
        "database.password" : "postgrespw",
        "database.dbname" : "orderdb",
        "database.server.name" : "dbserver1",
        "schema.whitelist" : "inventory",
        "table.whitelist" : "inventory.outboxevent",
        "tombstones.on.delete" : "false",
        "transforms" : "router",
        "transforms.router.type" : "io.debezium.examples.outbox.routingsmt.EventRouter"
    }
}

这将设置 io.debezium.connector.postgresql.PostgresConnector 的一个实例，以捕获来自指定 Postgres 实例的更改。请注意，通过表白名单，只会捕获来自 outboxevent 表的更改。它还应用了名为 EventRouter 的单消息转换 (SMT)。

从Kafka主题中删除事件通过将tombstones.on.delete设置为false，连接器在从信箱表中删除事件记录时不会发出删除标记（“tombstones”）。这是有意义的，因为从信箱表中删除不应影响相应Kafka主题中事件的保留。相反，可以在Kafka中配置事件主题的特定保留时间，例如保留所有采购订单事件30天。

或者，可以使用物理删除。这将需要对信箱表中事件的设计进行一些更改：

• 它们必须描述整个聚合；因此，例如，表示取消单个订单行的事件也应描述包含采购订单的完整当前状态；这样，消费者将能够在只看到与给定订单相关的最后一个事件时，通过日志压缩后仍能获取采购订单的完整状态。
• 它们必须有一个额外的布尔属性，指示特定事件是否表示事件的聚合根的删除。这样的事件（例如类型为OrderDeleted的事件）可以由下一节中描述的事件路由SMT使用，以生成该聚合根的删除标记。当OrderDeleted事件已写入主题时，日志压缩将删除与给定采购订单相关的所有事件。

当删除事件时，事件流将无法再从起始位置进行回放。根据特定的业务需求，仅保留给定采购订单、客户等的最终状态可能已足够。可以使用压缩的主题和足够的topic设置值来实现这一点。另一种选择是将历史事件移动到某种冷存储（例如Amazon S3存储桶），在需要时可以从中检索，然后从Kafka主题读取最新事件。选择哪种方法取决于具体的需求、预期的数据量以及开发和操作解决方案的团队的专业知识。

主题路由

默认情况下，Debezium连接器将将来自同一给定表的所有更改事件发送到同一个主题，即我们将得到一个名为dbserver1.inventory.outboxevent的单个Kafka主题，其中包含所有事件，无论是订单事件、客户事件等等。

为了简化仅对特定事件类型感兴趣的消费者的实现，更合理的做法是拥有多个主题，例如OrderEvents、CustomerEvents等。例如，发货服务可能对任何客户事件都不感兴趣。只订阅OrderEvents主题，就可以确保它永远不会收到任何客户事件。

为了将从信箱表中捕获的更改事件路由到不同的主题，使用了自定义的SMT EventRouter。以下是其apply()方法的代码，该方法将由Kafka Connect为Debezium连接器发出的每个记录调用：

@Override
public R apply(R record) {
    // Ignoring tombstones just in case
    if (record.value() == null) {
        return record;
    }

    Struct struct = (Struct) record.value();
    String op = struct.getString("op");

    // ignoring deletions in the outbox table
    if (op.equals("d")) {
        return null;
    }
    else if (op.equals("c")) {
        Long timestamp = struct.getInt64("ts_ms");
        Struct after = struct.getStruct("after");

        String key = after.getString("aggregateid");
        String topic = after.getString("aggregatetype") + "Events";

        String eventId = after.getString("id");
        String eventType = after.getString("type");
        String payload = after.getString("payload");

        Schema valueSchema = SchemaBuilder.struct()
            .field("eventType", after.schema().field("type").schema())
            .field("ts_ms", struct.schema().field("ts_ms").schema())
            .field("payload", after.schema().field("payload").schema())
            .build();

        Struct value = new Struct(valueSchema)
            .put("eventType", eventType)
            .put("ts_ms", timestamp)
            .put("payload", payload);

        Headers headers = record.headers();
        headers.addString("eventId", eventId);

        return record.newRecord(topic, null, Schema.STRING_SCHEMA, key, valueSchema, value,
                record.timestamp(), headers);
    }
    // not expecting update events, as the outbox table is "append only",
    // i.e. event records will never be updated
    else {
        throw new IllegalArgumentException("Record of unexpected op type: " + record);
    }
}

当接收到删除事件（op = d）时，它将丢弃该事件，因为从信箱表中删除事件记录对下游消费者来说并不重要。当接收到创建事件（op = c）时，情况变得更有趣。这样的记录将被传播到Apache Kafka。

Debezium的更改事件具有复杂的结构，包含所表示行的旧（before）状态和新（after）状态。要传播的事件结构是从after状态中获得的。从捕获的事件记录中获取的aggregatetype值用于构建将事件发送到的主题名称。例如，aggregatetype设置为Order的事件将被发送到OrderEvents主题。aggregateid用作消息键，确保该聚合的所有消息都将进入该主题的同一个分区。消息值是一个结构，包括原始事件负载（编码为JSON），指示事件生成时间的时间戳和事件类型。最后，事件UUID作为Kafka头字段进行传播。这使得消费者可以通过有效的重复检测，而无需检查实际的消息内容。

Apache Kafka 中的事件

现在让我们来看一下OrderEvents和CustomerEvents主题。

如果您已经检出了示例源代码并通过Docker Compose启动了所有组件（有关详细信息，请参阅示例项目中的README.md文件），您可以通过订单服务的REST API来下订单，如下所示：

cat resources/data/create-order-request.json | http POST http://localhost:8080/order-service/rest/orders

同样，可以取消特定的订单行：

cat resources/data/cancel-order-line-request.json | http PUT http://localhost:8080/order-service/rest/orders/1/lines/2

当使用非常实用的 kafkacat 实用程序等工具时，您现在应该在 OrderEvents 主题中看到如下消息：

kafkacat -b kafka:9092 -C -o beginning -f 'Headers: %h\nKey: %k\nValue: %s\n' -q -t OrderEvents

Headers: eventId=d03dfb18-8af8-464d-890b-09eb8b2dbbdd
Key: "4"
Value: {"eventType":"OrderCreated","ts_ms":1550307598558,"payload":"{\"id\": 4, \"lineItems\": [{\"id\": 7, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 8, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}"}
Headers: eventId=49f89ea0-b344-421f-b66f-c635d212f72c
Key: "4"
Value: {"eventType":"OrderLineUpdated","ts_ms":1550308226963,"payload":"{\"orderId\": 4, \"newStatus\": \"CANCELLED\", \"oldStatus\": \"ENTERED\", \"orderLineId\": 7}"}

payload字段是消息值的字符串化JSON表示形式。Debezium Postgres连接器将JSONB列作为字符串发出（使用io.debezium.data.Json逻辑类型名称），这就是为什么引号被转义的原因。jq实用程序，更具体地说，它的fromjson运算符，非常适合以更可读的方式显示事件负载：

kafkacat -b kafka:9092 -C -o beginning -t Order | jq '.payload | fromjson'

{
  "id": 4,
  "lineItems": [
    {
      "id": 7,
      "item": "Debezium in Action",
      "status": "ENTERED",
      "quantity": 2,
      "totalPrice": 39.98
    },
    {
      "id": 8,
      "item": "Debezium for Dummies",
      "status": "ENTERED",
      "quantity": 1,
      "totalPrice": 29.99
    }
  ],
  "orderDate": "2019-01-31T12:13:01",
  "customerId": 123
}
{
  "orderId": 4,
  "newStatus": "CANCELLED",
  "oldStatus": "ENTERED",
  "orderLineId": 7
}

您还可以查看 CustomerEvents 主题来检查添加采购订单时代表发票创建的事件。

消费服务中的重复检测

到目前为止，我们的信箱模式实现已经完全可用。当订单服务接收到下订单（或取消订单行）的请求时，它将在其数据库的purchaseorder和orderline表中持久化相应的状态。同时，在同一事务中，相应的事件条目将被添加到同一数据库的信箱表中。Debezium Postgres连接器捕获该表中的任何插入操作，并将事件路由到由给定事件表示的聚合类型对应的Kafka主题中。

为了总结一下，让我们来探讨另一个微服务（例如发货服务）如何消费这些消息。该服务的入口点是一个常规的Kafka消费者实现，在此为了简洁起见省略了。您可以在示例仓库中找到其source code。对于Order主题上的每个传入消息，消费者将调用OrderEventHandler。

@ApplicationScoped
public class OrderEventHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(OrderEventHandler.class);

    @Inject
    private MessageLog log;

    @Inject
    private ShipmentService shipmentService;

    @Transactional
    public void onOrderEvent(UUID eventId, String key, String event) {
        if (log.alreadyProcessed(eventId)) {
            LOGGER.info("Event with UUID {} was already retrieved, ignoring it", eventId);
            return;
        }

        JsonObject json = Json.createReader(new StringReader(event)).readObject();
        JsonObject payload = json.containsKey("schema") ? json.getJsonObject("payload") :json;

        String eventType = payload.getString("eventType");
        Long ts = payload.getJsonNumber("ts_ms").longValue();
        String eventPayload = payload.getString("payload");

        JsonReader payloadReader = Json.createReader(new StringReader(eventPayload));
        JsonObject payloadObject = payloadReader.readObject();

        if (eventType.equals("OrderCreated")) {
            shipmentService.orderCreated(payloadObject);
        }
        else if (eventType.equals("OrderLineUpdated")) {
            shipmentService.orderLineUpdated(payloadObject);
        }
        else {
            LOGGER.warn("Unkown event type");
        }

        log.processed(eventId);
    }
}

onOrderEvent()首先检查具有给定UUID的事件是否之前已经处理过。如果是这样，对于相同事件的任何进一步调用都将被忽略。这是为了防止由于数据管道的“至少一次”语义而导致的事件重复处理。例如，可能发生在Debezium连接器或消费服务在确认与源数据库或消息代理的特定事件检索之前失败的情况下。在这种情况下，在重新启动Debezium或消费服务后，可能会再次处理几个事件。将事件UUID作为Kafka消息头进行传播，可以有效地在消费者中检测和排除重复项。

如果第一次收到消息，将解析消息值，并使用特定事件类型对应的ShippingService方法的业务方法调用事件负载。最后，使用消息日志将消息标记为已处理。

这个MessageLog简单地在服务的本地数据库中的一个表中跟踪所有已消费的事件：

@ApplicationScoped
public class MessageLog {

    @PersistenceContext
    private EntityManager entityManager;

    @Transactional(value=TxType.MANDATORY)
    public void processed(UUID eventId) {
        entityManager.persist(new ConsumedMessage(eventId, Instant.now()));
    }

    @Transactional(value=TxType.MANDATORY)
    public boolean alreadyProcessed(UUID eventId) {
        return entityManager.find(ConsumedMessage.class, eventId) != null;
    }
}

这样，如果事务由于某种原因回滚，原始消息也不会被标记为已处理，并且异常将冒泡到Kafka事件消费者循环中。这允许稍后重试处理消息。

请注意，更完整的实现应该注意仅重试给定消息一定次数，然后将任何无法处理的消息重定向到死信队列或类似的队列。此外，还应对消息日志表进行一些维护；定期删除所有早于与代理提交的消费者当前偏移量的事件，因为保证这样的消息不会再次传播到消费者。

总结

信箱模式是在不同微服务之间传播数据的好方法。

通过仅修改单个资源 - 源服务自己的数据库 - 它避免了同时修改多个不共享一个常规事务上下文（数据库和Apache Kafka）的资源可能存在的任何不一致性。通过首先写入数据库，源服务具有即时的“读取自己的写入”语义，这对于一致的用户体验非常重要，允许在写入后立即调用的查询方法反映任何数据变化。

同时，该模式还实现了对其他微服务的异步事件传播。Apache Kafka作为服务之间消息传递的高度可伸缩和可靠的基础设施。在正确的主题保留设置下，新的消费者可以在事件最初产生之后很长时间才出现，并根据事件历史记录建立自己的本地状态。

将Apache Kafka放在整体架构的中心还确保了所涉及服务的解耦。例如，如果解决方案的单个组件失败或在某段时间内不可用（例如在更新期间），事件将简单地稍后处理：重新启动后，Debezium连接器将继续从之前离开的位置跟踪信箱表。同样，任何消费者将继续从其先前的偏移量处理主题。通过跟踪已成功处理的消息，可以检测到重复项并排除其重复处理。

当然，不同服务之间的此类事件管道最终是一致的，即运输服务等消费者可能会滞后于订单服务等生产者。通常情况下，这是可以接受的，并且可以根据应用程序的业务逻辑进行处理。例如，通常不需要在下订单的那一刻立即创建发货。此外，由于基于日志的变更数据捕获允许以近实时的方式发出事件，因此整体解决方案的端到端延迟通常很低（秒甚至亚秒范围）。

最后要记住的是，通过信箱公开的事件的结构应被视为发出服务的API的一部分。也就是说，当需要时，应该仔细调整它们的结构，并考虑到兼容性。这是为了确保在升级生产服务时不会意外破坏任何消费者。同时，消费者在处理消息时应该宽容一些，例如在接收到的事件中遇到未知属性时不要失败。