1
{
2
  "type": "record",
3
  "name": "OrderEvent",
4
  "namespace": "com.example.order",
5
  "fields": [
6
    {"name": "eventId", "type": "string"},
7
    {"name": "eventType", "type": {"type": "enum", "name": "OrderEventType", "symbols": ["CREATED", "PAID", "SHIPPED", "CANCELLED", "REFUNDED"]}},
8
    {"name": "orderId", "type": "string"},
9
    {"name": "customerId", "type": "string"},
10
    {"name": "amount", "type": "double"},
11
    {"name": "currency", "type": "string", "default": "CNY"},
12
    {"name": "items", "type": {"type": "array", "items": "string"}, "default": []},
13
    {"name": "timestamp", "type": "long"},
14
    {"name": "version", "type": "int", "default": 1},
15
    {"name": "shippingAddress", "type": ["null", "string"], "default": null},
16
    {"name": "priority", "type": "int", "default": 0}
17
  ]
18
}

1
@Service
2
public class OrderService {
3
    private final OrderRepository orderRepository;
4
    private final OutboxEventRepository outboxRepository;
5
    @Transactional
6
    public Order createOrder(CreateOrderRequest request) {
7
        // 1. 创建订单
8
        Order order = new Order();
9
        order.setId(UUID.randomUUID().toString());
10
        order.setCustomerId(request.getCustomerId());
11
        order.setAmount(request.getAmount());
12
        order.setStatus(OrderStatus.CREATED);
13
        order.setItems(request.getItems());
14
        orderRepository.save(order);
15
        // 2. 写入 Outbox（同一事务）
16
        OutboxEvent event = new OutboxEvent();
17
        event.setId(UUID.randomUUID().toString());
18
        event.setAggregateType("Order");
19
        event.setAggregateId(order.getId());
20
        event.setEventType("ORDER_CREATED");
21
        event.setPayload(JSON.toJSONString(order));
22
        outboxRepository.save(event);
23
        return order;
24
    }
25
    @Transactional
26
    public Order payOrder(String orderId, BigDecimal amount) {
27
        Order order = orderRepository.findById(orderId).orElseThrow();
28
        order.setStatus(OrderStatus.PAID);
29
        orderRepository.save(order);
30
        OutboxEvent event = new OutboxEvent();
31
        event.setId(UUID.randomUUID().toString());
32
        event.setAggregateType("Order");
33
        event.setAggregateId(orderId);
34
        event.setEventType("ORDER_PAID");
35
        event.setPayload(JSON.toJSONString(Map.of("orderId", orderId, "amount", amount)));
36
        outboxRepository.save(event);
37
        return order;
38
    }
39
}

1
{
2
  "name": "order-outbox-connector",
3
  "config": {
4
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
5
    "database.hostname": "mysql",
6
    "database.port": "3306",
7
    "database.user": "debezium",
8
    "database.password": "debezium",
9
    "database.server.id": "184054",
10
    "database.server.name": "order_service",
11
    "database.include.list": "order_db",
12
    "table.include.list": "order_db.outbox_events",
13
    "transforms": "outbox",
14
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
15
    "transforms.outbox.route.topic.replacement": "order-events",
16
    "transforms.outbox.table.field.event.id": "id",
17
    "transforms.outbox.table.field.event.key": "aggregateid",
18
    "transforms.outbox.table.field.event.type": "eventtype",
19
    "transforms.outbox.table.field.event.payload": "payload",
20
    "transforms.outbox.table.field.event.payload.id": "aggregateid"
21
  }
22
}

1
StreamsBuilder builder = new StreamsBuilder();
2
// 1. 读取订单事件
3
KStream<String, GenericRecord> orderEvents = builder.stream("order-events");
4
// 2. 按客户分组，5分钟翻滚窗口统计
5
KTable<Windowed<String>, OrderStats> windowedStats = orderEvents
6
    .groupBy((key, event) -> event.get("customerId").toString())
7
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).grace(Duration.ofMinutes(1)))
8
    .aggregate(
9
        OrderStats::new,
10
        (customerId, event, stats) -> {
11
            stats.setOrderCount(stats.getOrderCount() + 1);
12
            stats.setTotalAmount(stats.getTotalAmount() + (double) event.get("amount"));
13
            return stats;
14
        },
15
        Materialized.with(Serdes.String(), new JsonSerde<>(OrderStats.class))
16
    );
17
// 3. 输出统计结果
18
windowedStats.toStream()
19
    .map((windowedKey, stats) -> new KeyValue<>(windowedKey.key(), stats))
20
    .to("order-stats", Produced.with(Serdes.String(), new JsonSerde<>(OrderStats.class)));
21
// 4. 异常检测：5分钟内消费超过阈值
22
KStream<String, OrderAlert> alerts = windowedStats
23
    .toStream()
24
    .filter((windowedKey, stats) -> stats.getTotalAmount() > 50000)
25
    .map((windowedKey, stats) -> new KeyValue<>(
26
        windowedKey.key(),
27
        new OrderAlert("HIGH_SPENDING", windowedKey.key(), stats.getTotalAmount())
28
    ));
29
alerts.to("order-alerts", Produced.with(Serdes.String(), new JsonSerde<>(OrderAlert.class)));

1
// 订单聚合根：事件溯源
2
public class OrderAggregate {
3
    private String orderId;
4
    private String customerId;
5
    private double amount;
6
    private OrderStatus status;
7
    private List<String> items;
8
    private List<Object> pendingEvents = new ArrayList<>();
9
    private long version = 0;
10
    // 从事件流重建状态
11
    public static OrderAggregate fromEvents(List<OrderEvent> events) {
12
        OrderAggregate aggregate = new OrderAggregate();
13
        for (OrderEvent event : events) {
14
            aggregate.apply(event);
15
            aggregate.version++;
16
        }
17
        return aggregate;
18
    }
19
    // Command：创建订单
20
    public void createOrder(String orderId, String customerId, double amount, List<String> items) {
21
        if (status != null) throw new IllegalStateException("Order already exists");
22
        OrderCreatedEvent event = new OrderCreatedEvent(orderId, customerId, amount, items);
23
        pendingEvents.add(event);
24
        apply(event);
25
    }
26
    // Command：支付订单
27
    public void payOrder(double paidAmount) {
28
        if (status != OrderStatus.CREATED) throw new IllegalStateException("Invalid status for payment");
29
        if (paidAmount != amount) throw new IllegalArgumentException("Payment amount mismatch");
30
        OrderPaidEvent event = new OrderPaidEvent(orderId, paidAmount);
31
        pendingEvents.add(event);
32
        apply(event);
33
    }
34
    // Command：取消订单
35
    public void cancelOrder(String reason) {
36
        if (status == OrderStatus.SHIPPED) throw new IllegalStateException("Cannot cancel shipped order");
37
        OrderCancelledEvent event = new OrderCancelledEvent(orderId, reason);
38
        pendingEvents.add(event);
39
        apply(event);
40
    }
41
    // 事件应用（状态转移）
42
    private void apply(OrderEvent event) {
43
        switch (event) {
44
            case OrderCreatedEvent e -> {
45
                this.orderId = e.orderId();
46
                this.customerId = e.customerId();
47
                this.amount = e.amount();
48
                this.items = e.items();
49
                this.status = OrderStatus.CREATED;
50
            }
51
            case OrderPaidEvent e -> this.status = OrderStatus.PAID;
52
            case OrderCancelledEvent e -> this.status = OrderStatus.CANCELLED;
53
            case OrderShippedEvent e -> this.status = OrderStatus.SHIPPED;
54
        }
55
    }
56
    public List<Object> getPendingEvents() { return pendingEvents; }
57
    public void clearPendingEvents() { pendingEvents.clear(); }
58
}

1
@Service
2
public class EventSourcedOrderService {
3
    private final KafkaTemplate<String, String> kafkaTemplate;
4
    private final OrderEventRepository eventRepository;
5
    @Transactional
6
    public Order handleCommand(CreateOrderCommand cmd) {
7
        // 1. 从事件存储重建聚合根
8
        List<OrderEvent> history = eventRepository.findByOrderId(cmd.getOrderId());
9
        OrderAggregate aggregate = OrderAggregate.fromEvents(history);
10
        // 2. 执行命令（产生新事件）
11
        aggregate.createOrder(cmd.getOrderId(), cmd.getCustomerId(),
12
                              cmd.getAmount(), cmd.getItems());
13
        // 3. 持久化新事件到 Outbox（保证一致性）
14
        for (Object event : aggregate.getPendingEvents()) {
15
            OutboxEvent outbox = new OutboxEvent();
16
            outbox.setId(UUID.randomUUID().toString());
17
            outbox.setAggregateType("Order");
18
            outbox.setAggregateId(cmd.getOrderId());
19
            outbox.setEventType(event.getClass().getSimpleName());
20
            outbox.setPayload(JSON.toJSONString(event));
21
            outboxRepository.save(outbox);
22
        }
23
        aggregate.clearPendingEvents();
24
        return aggregate.toOrder();
25
    }
26
}

1
// MySQL 投影：构建订单查询视图
2
@Service
3
public class OrderProjection {
4
    private final JdbcTemplate jdbcTemplate;
5
    @KafkaListener(topics = "order-events", groupId = "order-projection-mysql")
6
    public void handleOrderEvent(ConsumerRecord<String, GenericRecord> record) {
7
        GenericRecord event = record.value();
8
        String eventType = event.get("eventType").toString();
9
        String orderId = event.get("orderId").toString();
10
        switch (eventType) {
11
            case "ORDER_CREATED" -> jdbcTemplate.update(
12
                "INSERT INTO order_view (id, customer_id, amount, status, items, created_at, updated_at) " +
13
                "VALUES (?, ?, ?, 'CREATED', ?, NOW(), NOW()) " +
14
                "ON DUPLICATE KEY UPDATE status='CREATED', updated_at=NOW()",
15
                orderId, event.get("customerId"), event.get("amount"), event.get("items").toString()
16
            );
17
            case "ORDER_PAID" -> jdbcTemplate.update(
18
                "UPDATE order_view SET status='PAID', updated_at=NOW() WHERE id=?",
19
                orderId
20
            );
21
            case "ORDER_SHIPPED" -> jdbcTemplate.update(
22
                "UPDATE order_view SET status='SHIPPED', updated_at=NOW() WHERE id=?",
23
                orderId
24
            );
25
            case "ORDER_CANCELLED" -> jdbcTemplate.update(
26
                "UPDATE order_view SET status='CANCELLED', updated_at=NOW() WHERE id=?",
27
                orderId
28
            );
29
        }
30
    }
31
}
32
// Elasticsearch 投影：构建搜索视图
33
@Service
34
public class OrderSearchProjection {
35
    private final RestHighLevelClient esClient;
36
    @KafkaListener(topics = "order-events", groupId = "order-projection-es")
37
    public void handleOrderEvent(ConsumerRecord<String, GenericRecord> record) {
38
        GenericRecord event = record.value();
39
        String orderId = event.get("orderId").toString();
40
        // 幂等更新：使用事件 ID 作为版本号
41
        UpdateRequest request = new UpdateRequest("orders", orderId);
42
        request.doc(JSON.toJSONString(event), XContentType.JSON);
43
        request.docAsUpsert(true);  // 不存在则插入
44
        request.retryOnConflict(3); // 版本冲突重试
45
        esClient.update(request, RequestOptions.DEFAULT);
46
    }
47
}

1
@Service
2
public class PaymentConsumer {
3
    @KafkaListener(topics = "order-events", groupId = "payment-service")
4
    public void handleOrderEvent(ConsumerRecord<String, GenericRecord> record) {
5
        GenericRecord event = record.value();
6
        String eventType = event.get("eventType").toString();
7
        String orderId = event.get("orderId").toString();
8
        switch (eventType) {
9
            case "ORDER_CREATED" -> {
10
                // 创建待支付记录
11
                Payment payment = new Payment();
12
                payment.setOrderId(orderId);
13
                payment.setAmount((double) event.get("amount"));
14
                payment.setStatus(PaymentStatus.PENDING);
15
                paymentRepository.save(payment);
16
            }
17
            case "ORDER_CANCELLED" -> {
18
                // 取消支付
19
                paymentRepository.cancelByOrderId(orderId);
20
            }
21
        }
22
    }
23
}

1
@Service
2
public class InventoryConsumer {
3
    @KafkaListener(topics = "order-events", groupId = "inventory-service")
4
    public void handleOrderEvent(ConsumerRecord<String, GenericRecord> record) {
5
        GenericRecord event = record.value();
6
        String eventType = event.get("eventType").toString();
7
        String orderId = event.get("orderId").toString();
8
        switch (eventType) {
9
            case "ORDER_CREATED" -> {
10
                // 锁定库存
11
                List<String> items = (List<String>) event.get("items");
12
                inventoryService.lockStock(items, orderId);
13
            }
14
            case "ORDER_PAID" -> {
15
                // 确认扣减库存
16
                inventoryService.confirmDeduct(orderId);
17
            }
18
            case "ORDER_CANCELLED" -> {
19
                // 释放库存
20
                inventoryService.releaseStock(orderId);
21
            }
22
        }
23
    }
24
}

1
version: "3.8"
2
services:
3
  mysql:
4
    image: mysql:8.0
5
    environment:
6
      MYSQL_ROOT_PASSWORD: root
7
      MYSQL_DATABASE: order_db
8
      MYSQL_USER: debezium
9
      MYSQL_PASSWORD: debezium
10
    command: >
11
      --server-id=1
12
      --log-bin=mysql-bin
13
      --binlog-format=ROW
14
      --binlog-row-image=FULL
15
      --expire-logs-days=7
16
    volumes:
17
      - mysql-data:/var/lib/mysql
18
    ports:
19
      - "3306:3306"
20
  zookeeper:
21
    image: confluentinc/cp-zookeeper:7.6.0
22
    environment:
23
      ZOOKEEPER_CLIENT_PORT: 2181
24
      ZOOKEEPER_TICK_TIME: 2000
25
    ports:
26
      - "2181:2181"
27
  kafka:
28
    image: confluentinc/cp-kafka:7.6.0
29
    depends_on: [zookeeper]
30
    environment:
31
      KAFKA_BROKER_ID: 1
32
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
33
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
34
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
35
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
36
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
37
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
38
      KAFKA_LOG_RETENTION_HOURS: 168
39
      KAFKA_NUM_PARTITIONS: 16
40
    volumes:
41
      - kafka-data:/var/lib/kafka/data
42
    ports:
43
      - "9092:9092"
44
  schema-registry:
45
    image: confluentinc/cp-schema-registry:7.6.0
46
    depends_on: [kafka]
47
    environment:
48
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
49
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
50
    ports:
51
      - "8081:8081"
52
  kafka-connect:
53
    image: confluentinc/cp-kafka-connect:7.6.0
54
    depends_on: [kafka, mysql]
55
    environment:
56
      CONNECT_BOOTSTRAP_SERVERS: kafka:9092
57
      CONNECT_REST_PORT: 8083
58
      CONNECT_GROUP_ID: order-connect-cluster
59
      CONNECT_CONFIG_STORAGE_TOPIC: connect-configs
60
      CONNECT_OFFSET_STORAGE_TOPIC: connect-offsets
61
      CONNECT_STATUS_STORAGE_TOPIC: connect-statuses
62
      CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
63
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
64
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
65
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
66
      CONNECT_PLUGIN_PATH: /usr/share/java,/etc/kafka-connect/jars
67
    volumes:
68
      - ./debezium-plugins:/etc/kafka-connect/jars
69
    ports:
70
      - "8083:8083"
71
  elasticsearch:
72
    image: elasticsearch:8.12.0
73
    environment:
74
      - discovery.type=single-node
75
      - xpack.security.enabled=false
76
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
77
    volumes:
78
      - es-data:/usr/share/elasticsearch/data
79
    ports:
80
      - "9200:9200"
81
  prometheus:
82
    image: prom/prometheus:v2.50.0
83
    volumes:
84
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
85
    ports:
86
      - "9090:9090"
87
  grafana:
88
    image: grafana/grafana:10.3.0
89
    depends_on: [prometheus]
90
    volumes:
91
      - ./monitoring/grafana-dashboards:/etc/grafana/provisioning/dashboards
92
    ports:
93
      - "3000:3000"
94
volumes:
95
  mysql-data:
96
  kafka-data:
97
  es-data:

1
groups:
2
  - name: order-system
3
    rules:
4
      - alert: ConsumerLagHigh
5
        expr: kafka_consumer_consumer_lag_records_lag_max > 100000
6
        for: 5m
7
        labels:
8
          severity: warning
9
        annotations:
10
          summary: "Consumer lag 超过 10 万条"
11
      - alert: DebeziumLagHigh
12
        expr: debezium_connector_source_lag_seconds > 5
13
        for: 2m
14
        labels:
15
          severity: critical
16
        annotations:
17
          summary: "Debezium 延迟超过 5 秒"
18
      - alert: SchemaIncompatible
19
        expr: schema_registry_compatibility_check_failed > 0
20
        for: 1m
21
        labels:
22
          severity: critical
23
        annotations:
24
          summary: "Schema 不兼容变更"

1
class OrderAggregateTest {
2
    @Test
3
    void shouldCreateOrder() {
4
        OrderAggregate aggregate = new OrderAggregate();
5
        aggregate.createOrder("ORD-1", "CUST-1", 100.0, List.of("ITEM-1"));
6
        assertEquals(OrderStatus.CREATED, aggregate.getStatus());
7
        assertEquals(1, aggregate.getPendingEvents().size());
8
        assertTrue(aggregate.getPendingEvents().get(0) instanceof OrderCreatedEvent);
9
    }
10
    @Test
11
    void shouldRebuildFromEvents() {
12
        List<OrderEvent> events = List.of(
13
            new OrderCreatedEvent("ORD-1", "CUST-1", 100.0, List.of("ITEM-1")),
14
            new OrderPaidEvent("ORD-1", 100.0)
15
        );
16
        OrderAggregate aggregate = OrderAggregate.fromEvents(events);
17
        assertEquals(OrderStatus.PAID, aggregate.getStatus());
18
        assertEquals(0, aggregate.getPendingEvents().size());  // 重建不产生新事件
19
    }
20
    @Test
21
    void shouldRejectDuplicatePayment() {
22
        OrderAggregate aggregate = new OrderAggregate();
23
        aggregate.createOrder("ORD-1", "CUST-1", 100.0, List.of("ITEM-1"));
24
        aggregate.payOrder(100.0);
25
        assertThrows(IllegalStateException.class, () -> aggregate.payOrder(100.0));
26
    }
27
}

1
@SpringBootTest
2
@EmbeddedKafka(partitions = 1, topics = {"order-events"})
3
class OrderProjectionTest {
4
    @Autowired
5
    private EmbeddedKafkaBroker embeddedKafka;
6
    @Autowired
7
    private KafkaTemplate<String, String> kafkaTemplate;
8
    @Test
9
    void shouldProjectOrderCreatedEvent() throws Exception {
10
        // 发送 ORDER_CREATED 事件
11
        String event = JSON.toJSONString(Map.of(
12
            "eventType", "ORDER_CREATED",
13
            "orderId", "ORD-1",
14
            "customerId", "CUST-1",
15
            "amount", 100.0
16
        ));
17
        kafkaTemplate.send("order-events", "ORD-1", event).get();
18
        // 等待投影更新
19
        Thread.sleep(2000);
20
        // 验证读模型
21
        OrderView view = orderViewRepository.findById("ORD-1").orElseThrow();
22
        assertEquals("CUST-1", view.getCustomerId());
23
        assertEquals(OrderStatus.CREATED, view.getStatus());
24
    }
25
}

1
@Testcontainers
2
class OrderSystemE2ETest {
3
    @Container
4
    static final KafkaContainer kafka = new KafkaContainer(
5
        DockerImageName.parse("confluentinc/cp-kafka:7.6.0")
6
    );
7
    @Container
8
    static final MySQLContainer<?> mysql = new MySQLContainer<>(
9
        DockerImageName.parse("mysql:8.0")
10
    ).withCommand("--binlog-format=ROW");
11
    @Test
12
    void shouldCompleteOrderFlow() {
13
        // 1. 创建订单 → 验证事件发送
14
        Order order = orderService.createOrder(request);
15
        await().atMost(5, SECONDS).until(() ->
16
            kafkaConsumer.records("order-events").size() >= 1
17
        );
18
        // 2. 支付订单 → 验证库存扣减
19
        orderService.payOrder(order.getId(), order.getAmount());
20
        await().atMost(5, SECONDS).until(() ->
21
            inventoryService.getLockedQuantity(order.getId()) > 0
22
        );
23
        // 3. 验证通知发送
24
        await().atMost(5, SECONDS).until(() ->
25
            notificationService.getSentCount(order.getId()) >= 1
26
        );
27
    }
28
}