DDD实践：在SpringBoot中跨微服务通过发件箱模式实现分布式事务机制 - Hans-Peter Grahsl

<b>public</b> <b>interface</b> Outboxable {

    <font><i>/**
     * DDD聚合Id，The id of the aggregate affected by a given event. This is also used to ensure strict
     * ordering of events belonging to one and the same aggregate.
     */</i></font><font>
    String getAggregateId();

    </font><font><i>/**
     * The type of the aggregate affected by a given event. This needs to be the same type string for all
     * related parts of one an the same aggregate that might get changed.
     */</i></font><font>
    String getAggregateType();

    </font><font><i>/**
     * The actual event payload as String e.g. JSON serialization.
     */</i></font><font>
    String getPayload();

    </font><font><i>/**
     * The (sub)type of an actual event which causes any changes to a specific aggregate type.
     */</i></font><font>
    String getType();

    </font><font><i>/**
     * The application timestamp at which the event has happened.
     */</i></font><font>
    Long getTimestamp();
}
</font>

@Entity
<b>public</b> <b>class</b> OutboxEvent {

    @Id
    @GeneratedValue
    @Type(type = <font>"uuid-char"</font><font>)
    <b>private</b> UUID id;

    @NotNull
    <b>private</b> String aggregateType;

    @NotNull
    <b>private</b> String aggregateId;

    @NotNull
    <b>private</b> String type;

    @NotNull
    <b>private</b> Long timestamp;

    @NotNull
    @Column(length = 1048576) </font><font><i>//e.g. 1 MB max</i></font><font>
    <b>private</b> String payload;

    <b>private</b> OutboxEvent() {
    }

    </font><font><i>//...</i></font><font>
}
</font>

@Component
<b>public</b> <b>class</b> OutboxListener {

    <b>private</b> OutboxEventRepository repository;

    <b>public</b> OutboxListener(OutboxEventRepository repository) {
        <b>this</b>.repository = repository;
    }

    @EventListener
    <b>public</b> <b>void</b> onExportedEvent(Outboxable event) {

        OutboxEvent outboxEvent = OutboxEvent.from(event);

        <font><i>// The outbox event will be written to the "outbox" table</i></font><font>
        </font><font><i>// and immediately afterwards removed again. Thus the</i></font><font>
        </font><font><i>// "outbox" table is effectively empty all the time. From a</i></font><font>
        </font><font><i>// CDC perspective this will produce an INSERT operation</i></font><font>
        </font><font><i>// followed by a DELETE operation of the same record such that</i></font><font>
        </font><font><i>// both events are captured from the database log by Debezium.</i></font><font>
        repository.save(outboxEvent);
        repository.delete(outboxEvent);

    }

}
</font>

@Entity
<b>public</b> <b>class</b> PurchaseOrder {

    <font><i>//...</i></font><font>

    @DomainEvents
    <b>private</b> Collection<Outboxable> triggerOutboxEvents() {
        <b>return</b> Arrays.asList(OrderUpsertedEvent.of(<b>this</b>));
    }

    </font><font><i>//...</i></font><font>

}
</font>

<b>public</b> <b>class</b> OrderUpsertedEvent implements Outboxable {

    <b>private</b> <b>static</b> ObjectMapper MAPPER = <b>new</b> ObjectMapper();

    <b>private</b> <b>final</b> Long id;
    <b>private</b> <b>final</b> JsonNode payload;
    <b>private</b> <b>final</b> Long timestamp;

    <b>static</b> {
        MAPPER.registerModule(<b>new</b> JavaTimeModule());
    }

    <b>private</b> OrderUpsertedEvent(Long id, JsonNode payload) {
        <b>this</b>.id = id;
        <b>this</b>.payload = payload;
        <b>this</b>.timestamp = Instant.now().getEpochSecond();
    }

    <b>public</b> <b>static</b> OrderUpsertedEvent of(PurchaseOrder order) {
        <b>return</b> <b>new</b> OrderUpsertedEvent(order.getId(), MAPPER.valueToTree(order));
    }

    @Override
    <b>public</b> String getAggregateId() {
        <b>return</b> String.valueOf(id);
    }

    @Override
    <b>public</b> String getAggregateType() {
        <b>return</b> PurchaseOrder.<b>class</b>.getName();
    }

    @Override
    <b>public</b> String getType() {
        <b>return</b> <b>this</b>.getClass().getName();
    }

    @Override
    <b>public</b> Long getTimestamp() {
        <b>return</b> timestamp;
    }

    @Override
    <b>public</b> String getPayload() {
        <b>try</b> {
            <b>return</b> MAPPER.writeValueAsString(payload);
        } <b>catch</b> (JsonProcessingException e) {
            e.printStackTrace();
        }
        <b>return</b> <b>null</b>;
    }

}

@Service
<b>public</b> <b>class</b> OrderService {

  <font><i>//...</i></font><font>

  @Transactional
  <b>public</b> PurchaseOrder placeOrder(PurchaseOrder order) {
    repository.save(order); </font><font><i>//NOTE: OrderUpsertedEvent automatically published behind the scenes</i></font><font>
    <b>return</b> order;
  }

  </font><font><i>//...</i></font><font>

}
</font>

@Service
<b>public</b> <b>class</b> OrderService {

    <font><i>//... </i></font><font>

    @Transactional
    <b>public</b> PurchaseOrder updateOrderLineStatus(<b>long</b> orderId, <b>long</b> orderLineId, OrderLineStatus newStatus) {
        PurchaseOrder po = repository.findById(orderId)
                .orElseThrow(() -> <b>new</b> EntityNotFoundException(</font><font>"order with id "</font><font> + orderId + </font><font>" doesn't exist!"</font><font>));
        OrderLineStatus oldStatus = po.updateOrderLine(orderLineId, newStatus);
        eventBus.publishEvent(OrderLineUpdatedEvent.of(orderId, orderLineId, newStatus, oldStatus));
        repository.save(po);
        <b>return</b> po;
    }

    </font><font><i>//...</i></font><font>

}
</font>

{
  “ name ”： “ mysql-outbox-src-connector-01 ”，
  “ config ”：{
    “ connector.<b>class</b> ”： “ io.debezium.connector.mysql.MySqlConnector ”，
    “ tasks.max ”： “ 1 ”，
    “ database.hostname ”： “ localhost ”，
    “ database.port ”： “ 3306 ”，
    “ database.user ”： “ debezium ”，
    “ database.password ”： “ dbz ”，
    “ database.server.id ”： “ 12345 ”，
    “ database.server.name ”： “ dbserver1 ”，
    “ database.whitelist ”： “ outbox-demo ”，
    “ database.serverTimezone ”： “欧洲/维也纳”，
    “ table.whitelist ”： “ outbox-demo.outbox_event ”，
    “ database.history.kafka.bootstrap.servers ”： “ localhost：9092 ”，
    “ database.history.kafka.topic ”： “ schema-changes.outbox-demo ”，
    “ tombstones.on.delete ”： “ false ”
  }
}

bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic dbserver1.outbox-demo.outbox_event --from-beginning | jq

{
  “ name ”： “ mdb-sink-outbox-raw ”，
  “ config ”：{
    “ key.converter ”： “ io.confluent.connect.avro.AvroConverter ”，
    “ key.converter.schema.registry.url ”： “ http：<font><i>// localhost：8081 ”，</i></font><font>
    “ value.converter ”： “ io.confluent.connect.avro.AvroConverter ”，
    “ value.converter.schema.registry.url ”： “ http：</font><font><i>// localhost：8081 ”，</i></font><font>
    “ connector.<b>class</b> ”： “ at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector ”，
    “ topics ”： “ dbserver1.outbox-demo.outbox_event ”，
    “ mongodb.connection.uri ”： “ mongodb：</font><font><i>// localhost：27017 / outboxed ”，</i></font><font>
    “ mongodb.collections ”： “发件箱，原料”，
    “ mongodb.collection.dbserver1.outbox-demo.outbox_event ”： “ outbox -raw ”，
    “ mongodb.change.data.capture.handler.outbox-raw ”： “ at.grahsl.kafka.connect.mongodb.cdc.debezium.rdbms.RdbmsHandler ”，
    “ mongodb.change.data.capture.handler.operations.outbox-raw ”： “ c ”
  }
}
</font>