1
// 事件定义
2
public interface DomainEvent {
3
    String getAggregateId();
4
    LocalDateTime getTimestamp();
5
}
6

7
public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount, LocalDateTime timestamp) implements DomainEvent {}
8
public record OrderPaidEvent(String orderId, BigDecimal paidAmount, LocalDateTime timestamp) implements DomainEvent {}
9
public record OrderShippedEvent(String orderId, String trackingNumber, LocalDateTime timestamp) implements DomainEvent {}
10
public record OrderCancelledEvent(String orderId, String reason, LocalDateTime timestamp) implements DomainEvent {}
11

12
// Aggregate：订单聚合根
13
public class OrderAggregate {
14
    private String orderId;
15
    private String customerId;
16
    private BigDecimal amount;
17
    private OrderStatus status;
18
    private final List<DomainEvent> uncommittedEvents = new ArrayList<>();
19

20
    // 从事件重建状态
21
    public static OrderAggregate fromEvents(List<DomainEvent> events) {
22
        OrderAggregate order = new OrderAggregate();
23
        events.forEach(order::apply);
24
        return order;
25
    }
26

27
    // 业务操作：创建订单
28
    public void create(String orderId, String customerId, BigDecimal amount) {
29
        if (status != null) throw new IllegalStateException("Order already exists");
30
        var event = new OrderCreatedEvent(orderId, customerId, amount, LocalDateTime.now());
31
        apply(event);
32
        uncommittedEvents.add(event);
33
    }
34

35
    // 业务操作：支付
36
    public void pay(BigDecimal paidAmount) {
37
        if (status != OrderStatus.CREATED) throw new IllegalStateException("Cannot pay: " + status);
38
        var event = new OrderPaidEvent(orderId, paidAmount, LocalDateTime.now());
39
        apply(event);
40
        uncommittedEvents.add(event);
41
    }
42

43
    // 应用事件（更新状态）
44
    private void apply(DomainEvent event) {
45
        switch (event) {
46
            case OrderCreatedEvent e -> {
47
                this.orderId = e.orderId();
48
                this.customerId = e.customerId();
49
                this.amount = e.amount();
50
                this.status = OrderStatus.CREATED;
51
            }
52
            case OrderPaidEvent e -> this.status = OrderStatus.PAID;
53
            case OrderShippedEvent e -> this.status = OrderStatus.SHIPPED;
54
            case OrderCancelledEvent e -> this.status = OrderStatus.CANCELLED;
55
        }
56
    }
57
}

1
// Event Store 接口
2
public interface EventStore {
3
    void append(String aggregateId, List<DomainEvent> events, long expectedVersion);
4
    List<DomainEvent> load(String aggregateId);
5
    List<DomainEvent> loadFromVersion(String aggregateId, long fromVersion);
6
}
7

8
// 基于 Kafka 的 Event Store
9
public class KafkaEventStore implements EventStore {
10
    private final KafkaTemplate<String, String> kafkaTemplate;
11
    private final ObjectMapper objectMapper;
12

13
    @Override
14
    public void append(String aggregateId, List<DomainEvent> events, long expectedVersion) {
15
        // 乐观并发控制：检查版本号
16
        List<DomainEvent> existing = load(aggregateId);
17
        if (existing.size() != expectedVersion) {
18
            throw new ConcurrentModificationException(
19
                "Expected version " + expectedVersion + " but was " + existing.size());
20
        }
21

22
        // 追加事件到 Kafka
23
        for (DomainEvent event : events) {
24
            ProducerRecord<String, String> record = new ProducerRecord<>(
25
                "events-" + aggregateId,  // 按 Aggregate 分区
26
                aggregateId,
27
                objectMapper.writeValueAsString(event)
28
            );
29
            kafkaTemplate.send(record).get();  // 同步发送确保持久化
30
        }
31
    }
32

33
    @Override
34
    public List<DomainEvent> load(String aggregateId) {
35
        // 从 Kafka 读取所有事件
36
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
37
        return records.records(new TopicPartition("events", partitionOf(aggregateId)))
38
            .stream()
39
            .map(this::deserializeEvent)
40
            .toList();
41
    }
42
}

1
// Command 定义
2
public record CreateOrderCommand(String orderId, String customerId, BigDecimal amount) {}
3
public record PayOrderCommand(String orderId, BigDecimal amount) {}
4
public record ShipOrderCommand(String orderId, String trackingNumber) {}
5
public record CancelOrderCommand(String orderId, String reason) {}
6

7
// Command Handler
8
@Service
9
public class OrderCommandHandler {
10
    private final EventStore eventStore;
11

12
    @Transactional
13
    public void handle(CreateOrderCommand cmd) {
14
        OrderAggregate order = new OrderAggregate();
15
        order.create(cmd.orderId(), cmd.customerId(), cmd.amount());
16
        eventStore.append(cmd.orderId(), order.getUncommittedEvents(), 0);
17
        order.markEventsAsCommitted();
18
    }
19

20
    @Transactional
21
    public void handle(PayOrderCommand cmd) {
22
        List<DomainEvent> events = eventStore.load(cmd.orderId());
23
        OrderAggregate order = OrderAggregate.fromEvents(events);
24
        order.pay(cmd.amount());
25
        eventStore.append(cmd.orderId(), order.getUncommittedEvents(), events.size());
26
        order.markEventsAsCommitted();
27
    }
28
}

1
// 事件投影：将事件转换为读模型
2
@Component
3
public class OrderProjection {
4

5
    @KafkaListener(topics = "order-events")
6
    public void handleEvent(ConsumerRecord<String, String> record) {
7
        DomainEvent event = deserializeEvent(record.value());
8

9
        switch (event) {
10
            case OrderCreatedEvent e -> {
11
                OrderReadModel readModel = new OrderReadModel();
12
                readModel.setOrderId(e.orderId());
13
                readModel.setCustomerId(e.customerId());
14
                readModel.setAmount(e.amount());
15
                readModel.setStatus("CREATED");
16
                readModel.setCreatedAt(e.timestamp());
17
                orderReadRepository.save(readModel);
18
            }
19
            case OrderPaidEvent e -> {
20
                OrderReadModel readModel = orderReadRepository.findById(e.orderId()).orElseThrow();
21
                readModel.setStatus("PAID");
22
                readModel.setPaidAt(e.timestamp());
23
                orderReadRepository.save(readModel);
24
            }
25
            case OrderShippedEvent e -> {
26
                OrderReadModel readModel = orderReadRepository.findById(e.orderId()).orElseThrow();
27
                readModel.setStatus("SHIPPED");
28
                readModel.setTrackingNumber(e.trackingNumber());
29
                orderReadRepository.save(readModel);
30
            }
31
            case OrderCancelledEvent e -> {
32
                OrderReadModel readModel = orderReadRepository.findById(e.orderId()).orElseThrow();
33
                readModel.setStatus("CANCELLED");
34
                readModel.setCancelReason(e.reason());
35
                orderReadRepository.save(readModel);
36
            }
37
        }
38
    }
39
}
40

41
// Query Service：直接查询读模型
42
@Service
43
public class OrderQueryService {
44
    public OrderReadModel getOrder(String orderId) {
45
        return orderReadRepository.findById(orderId).orElseThrow();
46
    }
47

48
    public List<OrderReadModel> getOrdersByCustomer(String customerId) {
49
        return orderReadRepository.findByCustomerId(customerId);
50
    }
51

52
    public Page<OrderReadModel> searchOrders(OrderSearchCriteria criteria) {
53
        return orderReadRepository.search(criteria);
54
    }
55
}

1
// 快照实现
2
public class SnapshotStore {
3
    private static final int SNAPSHOT_INTERVAL = 1000;  // 每 1000 个事件一个快照
4

5
    public void maybeSaveSnapshot(String aggregateId, List<DomainEvent> allEvents) {
6
        if (allEvents.size() % SNAPSHOT_INTERVAL == 0) {
7
            OrderAggregate aggregate = OrderAggregate.fromEvents(allEvents);
8
            Snapshot snapshot = new Snapshot(
9
                aggregateId,
10
                allEvents.size(),
11
                aggregate.getState(),
12
                LocalDateTime.now()
13
            );
14
            snapshotRepository.save(snapshot);
15
        }
16
    }
17

18
    public OrderAggregate loadWithSnapshot(String aggregateId) {
19
        // 1. 加载最近的快照
20
        Snapshot snapshot = snapshotRepository.findLatest(aggregateId);
21

22
        // 2. 加载快照之后的事件
23
        List<DomainEvent> events;
24
        if (snapshot != null) {
25
            events = eventStore.loadFromVersion(aggregateId, snapshot.getVersion());
26
            OrderAggregate aggregate = OrderAggregate.fromSnapshot(snapshot);
27
            events.forEach(aggregate::apply);
28
            return aggregate;
29
        } else {
30
            events = eventStore.load(aggregateId);
31
            return OrderAggregate.fromEvents(events);
32
        }
33
    }
34
}

1
// 处理最终一致性的策略
2
@RestController
3
@RequestMapping("/api/orders")
4
public class OrderController {
5

6
    @PostMapping
7
    public ResponseEntity<OrderResponse> createOrder(@RequestBody CreateOrderRequest req) {
8
        // 写入
9
        commandHandler.handle(new CreateOrderCommand(req.getOrderId(), req.getCustomerId(), req.getAmount()));
10

11
        // 策略一：返回 202 Accepted，客户端轮询
12
        return ResponseEntity.accepted()
13
            .body(new OrderResponse(req.getOrderId(), "PENDING", "Order is being processed"));
14

15
        // 策略二：等待读模型更新（带超时）
16
        // awaitReadModelUpdate(req.getOrderId(), Duration.ofSeconds(5));
17
    }
18
}

1
// 事件升级器：将旧版本事件转换为新版本
2
public class EventUpgrader {
3
    public DomainEvent upgrade(DomainEvent event) {
4
        return switch (event) {
5
            case OrderCreatedEventV1 v1 -> upgradeV1ToV2(v1);
6
            case OrderCreatedEventV2 v2 -> v2;  // 最新版本，无需升级
7
            default -> event;
8
        };
9
    }
10

11
    private OrderCreatedEventV2 upgradeV1ToV2(OrderCreatedEventV1 v1) {
12
        // V1 没有 currency 字段，V2 新增了 currency
13
        return new OrderCreatedEventV2(
14
            v1.orderId(), v1.customerId(), v1.amount(),
15
            "CNY",  // 默认值
16
            v1.timestamp()
17
        );
18
    }
19
}

1
-- 事件存储表的典型 Schema
2
CREATE TABLE event_store (
3
    event_id      UUID PRIMARY KEY,
4
    aggregate_id  UUID NOT NULL,
5
    event_type    VARCHAR(100) NOT NULL,
6
    event_data    JSONB NOT NULL,
7
    version       BIGINT NOT NULL,
8
    timestamp     TIMESTAMPTZ DEFAULT NOW(),
9
    UNIQUE (aggregate_id, version)  -- 乐观锁：同一聚合版本唯一
10
);
11

12
-- 查询某个聚合的完整事件历史
13
SELECT event_type, event_data, version, timestamp
14
FROM event_store
15
WHERE aggregate_id = 'order-12345'
16
ORDER BY version ASC;
17

18
-- 查询某类事件（投影构建）
19
SELECT event_data->>'product_id' AS product_id,
20
       COUNT(*) AS order_count,
21
       SUM((event_data->>'amount')::numeric) AS total_amount
22
FROM event_store
23
WHERE event_type = 'OrderPlaced'
24
  AND timestamp >= NOW() - INTERVAL '7 days'
25
GROUP BY event_data->>'product_id';