Outbox Quarkus 扩展

概述

本扩展的灵感来源于 Outbox Event Router 单消息转换（SMT）。正如博文使用 Outbox 模式实现可靠的微服务数据交换中所讨论的，微服务之间经常需要交换信息，而应对这一需求的绝佳方式，就是将 Outbox 模式与 Debezium 的 Outbox Event Router SMT 结合使用。

下图展示了该模式的整体架构：

Outbox 模式

Outbox 扩展的目标是为 Quarkus 应用提供一个可复用、高度可配置的组件，该组件便于配合 Debezium 的 CDC 连接器管道使用 Outbox 模式，从而可靠、异步地将数据共享给该数据的任意使用方。

快速开始

要开始使用 Debezium Outbox Quarkus 扩展，需要将该扩展作为 Quarkus 应用的一部分添加进来，方式如下：

<dependency>
  <groupId>io.debezium</groupId>
  <artifactId>debezium-quarkus-outbox</artifactId>
  <version>3.6.3.Final</version>
</dependency>

该扩展为应用提供了 io.debezium.outbox.quarkus.ExportedEvent 接口。应用类应当实现该接口，并通过 javax.enterprise.event.Event 类来发送事件。

	`ExportedEvent` 接口带有类型参数，以便应用为事件发出的若干属性指定所使用的 Java 类型。需要注意的是，对于给定的 Quarkus 应用，所有 `ExportedEvent` 接口的实现都必须使用相同的参数类型，否则构建将失败，因为所有事件都会使用同一个底层数据库表。

示例

以下示例展示了 ExportedEvent 接口的一个实现，表示一个已创建的订单：

OrderCreatedEvent.java

public class OrderCreatedEvent implements ExportedEvent<String, JsonNode> {

    private static final String TYPE = "Order";
    private static final String EVENT_TYPE = "OrderCreated";

    private final long orderId;
    private final JsonNode jsonNode;
    private final Instant timestamp;

    public OrderCreatedEvent(Instant createdAt, Order order) {
        this.orderId = order.getId();
        this.timestamp = createdAt;
        this.jsonNode = convertToJson(order);
    }

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

    @Override
    public String getAggregateType() {
        return TYPE;
    }

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

    @Override
    public String getType() {
        return EVENT_TYPE;
    }

    @Override
    public Instant getTimestamp() {
        return timestamp;
    }

    @Override
    public Map<String, Object> getAdditionalFieldValues() {
        // no additional fields
        return Collections.emptyMap();
    }
}

下面的示例展示了一个发出 OrderCreatedEvent 的 OrderService：

OrderService.java

@ApplicationScoped
public class OrderService {

    @Inject
    OrderRepository orderRepository;

    @Inject
    Event<ExportedEvent<?, ?>> event;

    @Transactional
    public Order addOrder(Order order) {
        order = orderRepository.save(order);
        event.fire(new OrderCreatedEvent(Instant.now(), order));
        return order;
    }
}

当应用代码调用 Event#fire() 触发事件时，Outbox 扩展会收到事件发生的事件通知，并在当前事务范围内将事件内容持久化到发件箱事件表中。Debezium CDC 连接器与 Outbox 事件路由器配合，将监控该表，并负责通过 CDC 事件转发这些数据。

如需查看完整的端到端演示，Outbox 示例展示了两个使用发件箱模式的 Quarkus 微服务应用，它们在下单或取消订单时相互共享数据。

响应式变体

如果你的应用使用响应式数据源或 Hibernate Reactive，则必须使用稍有不同的配置来将该扩展添加到应用中。

例如，使用以下配置导入该扩展的响应式变体：

<dependency>
  <groupId>io.debezium</groupId>
  <artifactId>debezium-quarkus-outbox-reactive</artifactId>
  <version>3.6.3.Final</version>
</dependency>

反应式扩展为应用提供了与非反应式变体相同的 io.debezium.outbox.quarkus.ExportedEvent 接口，此外还提供了 DebeziumOutboxHandler 类。将 DebeziumOutboxHandler Bean 注入到应用中，即可通过一种快捷方式将 ExportedEvent 持久化到发件箱表。下面的示例展示了如何调用，其中使用了与前一个示例相同的 ExportedEvent：

OrderService.java

@ApplicationScoped
public class OrderService {

    @Inject
    DebeziumOutboxHandler debeziumOutboxHandler;

    @Inject
    OrderRepository orderRepository;

    @Inject
    Event<ExportedEvent<?, ?>> event;

    @ReactiveTransactional
    public Uni<Order> addOrder(Order order) {
        return orderRepository.persistAndFlush(order)
        .call(debeziumOutboxHandler.persistToOutbox(new OrderCreatedEvent(Instant.now(), order)));

    }
}

	`persistToOutbox` 方法返回一个 Uni，因此必须由订阅者进行观察才能接收结果。有关使用 Mutiny 库构建响应式应用的更多信息，请参阅 Quarkus Mutiny 入门指南。

附加字段映射

Debezium Outbox SMT 可以配置为读取附加字段，并将这些字段值作为事件头发出，或作为事件值的一部分发出。

为了向 Quarkus Outbox 扩展传递需要保存的附加字段映射，必须在 application.properties 中指定配置属性 quarkus.debezium-outbox.additional-fields。该配置属性是一个以逗号分隔的附加字段定义列表，这些定义将被添加到 Outbox 实体映射中，并由应用对 ExportedEvent 接口的实现传递。

此逗号分隔列表中的每个条目都必须遵循以下格式：

<field-name>:<field-java-type>[:<field-column-definition>[:<field-jpa-attribute-converter>]]

该模式表明字段的名称和 java-type 是必填的，而列定义和 JPA 属性转换器则是可选的。但请注意，如果要指定 JPA 属性转换器，则必须同时指定列定义。

下面的示例演示了如何定义一个名为 customer_name 的附加字段，该字段在 Java 中表示为 String，在 outbox 表中应存储为 VARCHAR(100) 列。此示例还展示了定义的一个 JPA 属性转换器，用于强制将该字符串以大写形式存储。

application.properties

quarkus.debezium-outbox.additional-fields=customer_name:string:varchar(100):example.UpperCase

在应用程序的 .properties 文件中配置好该字段之后，应用程序的代码需要通过其导出的事件提供相应的值。为此，继承 ExportedEvent 的应用程序类需要重写名为 getAdditionalFieldValues() 的方法，并返回一个包含额外字段名称及其取值的 Map。

下面的示例展示了如何指定 customer_name 字段，其值为 Acme Goods。沿用上文示例部分中的 OrderCreatedEvent，我们对该事件进行了扩展：

OrderCreatedEvent.java

public class OrderCreatedEvent implements ExportedEvent<String, JsonNode> {
    ...
    @Override
    public Map<String, Object> getAdditionalFieldValues() {
        return Collections.singletonMap("customer_name", "Acme Goods");
    }
}