Spring Boot 3.5 + Java 25 + Cloud Native 系列（五）：事件驱动架构

public record EventEnvelope<T>(
        String eventId,           // 事件唯一 ID（UUID）
        String source,            // 事件来源（如 "order-service"）
        String type,              // 事件类型（如 "ORDER_COMPLETED"）
        Instant timestamp,        // 发生时间
        Integer schemaVersion,    // Schema 版本号，默认 1
        String contentType,       // 内容类型，默认 "application/json"
        T data) {                 // 实际业务数据

    public static final int CURRENT_SCHEMA_VERSION = 1;

    public void assertSupportedSchema(int... supportedSchemaVersions) {
        // 验证 schemaVersion 是否在支持列表中
    }
}

// 生产者
EventEnvelope<OrderEventData> envelope = new EventEnvelope<>(
        UUID.randomUUID().toString(),
        "order-service",
        "ORDER_COMPLETED",
        Instant.now(),
        new OrderEventData(
                order.getId(),
                order.getOrderNo(),
                order.getBuyerId(),
                order.getSellerId(),
                order.getStatus(),
                order.getTotalAmount(),
                order.getItems().stream()
                    .map(item -> new OrderItemSummary(
                            item.getProductId(), item.getProductName(),
                            item.getQuantity(), item.getLineTotal()))
                    .toList()));

kafkaTemplate.send(topic, objectMapper.writeValueAsString(envelope));

// 消费者
@KafkaListener(topics = "${shop.loyalty.order-events-topic}")
public void onOrderEvent(String payload) {
    EventEnvelope<OrderEventData> envelope = objectMapper.readValue(
            payload, new TypeReference<>() {});
    envelope.assertSupportedSchema(1);
    OrderEventData data = envelope.data();
    // 处理业务逻辑
}

public record OrderEventData(
        String orderId, String orderNo, String buyerId,
        String sellerId, String status, BigDecimal totalAmount,
        List<OrderItemSummary> items) {
    public record OrderItemSummary(String productId, String productName,
                                   int quantity, BigDecimal lineTotal) {}
}

@Entity
@Table(name = "wallet_outbox_event")
public class WalletOutboxEventEntity {
    @Id
    private String id;  // UUID
    private String aggregateId;  // wallet_transaction_id
    private String topic;
    private String eventType;
    private String payload;  // EventEnvelope JSON
    private boolean published;
}

@Scheduled(fixedDelayString = "${shop.wallet.outbox-publish-delay-ms:5000}")
@Transactional
public void publishPendingEvents() {
    List<WalletOutboxEventEntity> events = repository
            .findTop20ByPublishedFalseOrderByCreatedAtAsc();
    for (WalletOutboxEventEntity event : events) {
        kafkaTemplate.send(event.getTopic(), event.getPayload());
        event.markPublished();
    }
    repository.saveAll(events);
}

// 使用 .join() 同步等待 Kafka 确认
kafkaTemplate.send(topic, aggregateId, payload).join();

// top20、无 key，发送后批量标记 published
kafkaTemplate.send(topic, payload);

// 查询 status='PENDING'，按 gameId 作为 key，逐条 save
kafkaTemplate.send(event.getTopic(), event.getGameId(), event.getPayload());

public interface IdempotencyGuard {
    <T> T executeOnce(String key, Supplier<T> action, Supplier<T> fallback);
}

public <T> T executeOnce(String key, Supplier<T> action, Supplier<T> fallback) {
    validateKey(key);
    if (repository.existsByKey(key)) {
        return fallback.get();  // 已处理过，走 fallback
    }
    try {
        return action.get();    // 执行业务
    } catch (DataIntegrityViolationException exception) {
        return fallback.get();  // 并发重复导致唯一键冲突
    }
}

public interface WalletIdempotencyKeyRepository
        extends JpaRepository<WalletIdempotencyKeyEntity, String>,
                IdempotencyRepository {
    boolean existsByIdempotencyKey(String idempotencyKey);
    @Override default boolean existsByKey(String key) {
        return existsByIdempotencyKey(key);
    }
}

public <T> T executeOnce(String key, Supplier<T> action, Supplier<T> fallback) {
    // Bloom Filter 快速检查
    if (!bloomFilter.contains(key)) {
        missCounter.increment();
        return executeAction(key, action, fallback);
    }

    // Bloom Filter 说"可能存在"→ 查 DB 确认
    if (repository.existsByKey(key)) {
        duplicateHitCounter.increment();
        return fallback.get();
    }

    // 误判（false positive），实际是新 key
    falsePositiveHitCounter.increment();
    return executeAction(key, action, fallback);
}

shop:
  idempotency:
    bloom-filter:
      enabled: true
      redis-key: "shop:idempotency:loyalty"
      expected-insertions: 1000000   # 预期插入量
      false-probability: 0.001       # 误判率 0.1%

@KafkaListener(topics = "${shop.loyalty.order-events-topic}")
@Transactional
public void onOrderEvent(String payload) {
    EventEnvelope<OrderEventData> envelope = objectMapper.readValue(
            payload, new TypeReference<>() {});
    OrderEventData data = envelope.data();

    String idempotencyKey = "LOYALTY_ORDER_COMPLETED:" + data.orderId();
    idempotencyGuard.executeOnce(
            idempotencyKey,
            () -> {
                handleOrderCompleted(data);
                // 在同一事务中保存幂等键
                idempotencyKeyRepository.save(
                        new LoyaltyIdempotencyKeyEntity(idempotencyKey));
                return null;
            },
            () -> {
                log.info("Loyalty order event already processed: orderId={}, skipping",
                        data.orderId());
                return null;
            });
}

@Component
@IdempotencyExempt(reason = "notification_log deduplicates by eventId + channel")
public class OrderEventListener {
    @KafkaListener(topics = "${shop.notification.order-events-topic}")
    public void onOrderEvent(String payload) {
        // 直接处理，notification_log 唯一约束承担去重
    }
}

@RetryableTopic(
        attempts = "4",                              // 最多尝试 4 次
        backoff = @Backoff(delay = 1000, multiplier = 2), // 指数退避
        dltTopicSuffix = ".dlq",                     // 死信话题后缀
        autoCreateTopics = "true",                   // 自动创建 DLQ topic
        exclude = NonRetryableKafkaConsumerException.class // 不可重试的异常
)
@KafkaListener(topics = "${shop.notification.order-events-topic}")
@Transactional
public void onOrderEvent(String payload) {
    // 处理逻辑
}

@DltHandler
public void handleDlt(String payload) {
    log.error("Event sent to DLT: {}", payload);
}

// 不可重试 → 直接送 DLQ
throw new NonRetryableKafkaConsumerException("Invalid JSON payload", exception);

// 可重试 → 走重试逻辑
throw new RetryableKafkaConsumerException("Database temporarily unavailable", exception);