event-driven-architecture

事件驱动架构

概述

此技能涵盖事件驱动系统的设计与实现，通过异步消息传递实现服务解耦。内容包括消息代理的选择与集成（Kafka、RabbitMQ、SQS/SNS、NATS）、事件溯源和 CQRS 模式、用于分布式事务的 Saga 编排、用于故障处理的死信队列、幂等模式、事件模式演进、事务发件箱模式，以及消费者组管理。

构建微服务架构、实现分布式工作流、解耦服务以便独立部署、处理最终一致性，或将服务间的同步调用替换为异步事件时，请使用此技能。

---

核心原则

1. 事件是事实，命令是请求 —— 事件描述已经发生的事情（OrderPlaced、PaymentReceived）。命令请求执行某项操作（ProcessPayment、ShipOrder）。这一区分是正确系统设计的基础：事件以广播方式发布，命令则点对点发送。
2. 幂等性不是可选项 —— 消息至少会投递一次（有时会投递多次）。每个消费者都必须妥善处理重复消息。使用事件 ID、版本检查或数据库约束进行去重。
3. 演进模式时不破坏消费者 —— 事件模式会发生变化。采用向后兼容的演进方式（添加字段，不删除或重命名字段）。明确标记模式版本，并在迁移期间支持多个版本。
4. 死信不是垃圾 —— 无法处理的消息会进入死信队列。这些消息可能反映了程序缺陷、数据问题或边缘情况。监控 DLQ，在数量增长时发出警报，并构建用于重新投递消息的工具。
5. 本地事务，最终一致性 —— 每个服务拥有自己的数据，并在本地事务中处理事件。跨服务的一致性是最终一致的，而非即时一致。设计 UI 和工作流时应考虑到这一点。

---

关键模式

模式 1：使用事务发件箱发布事件

适用场景： 需要原子性地更新数据库并发布事件时。这样可以避免“双写”问题，即数据库写入成功但事件发布失败（或反之）。

实现：

// The Transactional Outbox Pattern
// 1. Write the event to an outbox table in the SAME database transaction as the business data
// 2. A separate poller/CDC process reads the outbox and publishes to the message broker

interface OutboxEvent {
  id: string;
  aggregateType: string;
  aggregateId: string;
  eventType: string;
  payload: Record<string, unknown>;
  createdAt: Date;
  publishedAt: Date | null;
}

// Step 1: Business operation + outbox write in one transaction
async function placeOrder(order: CreateOrderInput): Promise<Order> {
  return await prisma.$transaction(async (tx) => {
    // Business logic
    const createdOrder = await tx.order.create({
      data: {
        customerId: order.customerId,
        items: { create: order.items },
        total: calculateTotal(order.items),
        status: "PLACED",
      },
      include: { items: true },
    });

    // Write event to outbox (same transaction!)
    await tx.outboxEvent.create({
      data: {
        id: crypto.randomUUID(),
        aggregateType: "Order",
        aggregateId: createdOrder.id,
        eventType: "OrderPlaced",
        payload: {
          orderId: createdOrder.id,
          customerId: createdOrder.customerId,
          items: createdOrder.items,
          total: createdOrder.total,
          placedAt: new Date().toISOString(),
        },
      },
    });

    return createdOrder;
  });
}

// Step 2: Outbox poller publishes events to Kafka/RabbitMQ
async function processOutbox(): Promise<void> {
  const unpublished = await prisma.outboxEvent.findMany({
    where: { publishedAt: null },
    orderBy: { createdAt: "asc" },
    take: 100,
  });

  for (const event of unpublished) {
    try {
      await kafka.producer.send({
        topic: `${event.aggregateType}.${event.eventType}`,
        messages: [
          {
            key: event.aggregateId,
            value: JSON.stringify({
              eventId: event.id,
              eventType: event.eventType,
              aggregateId: event.aggregateId,
              payload: event.payload,
              timestamp: event.createdAt.toISOString(),
            }),
            headers: {
              "event-type": event.eventType,
              "event-id": event.id,
            },
          },
        ],
      });

      await prisma.outboxEvent.update({
        where: { id: event.id },
        data: { publishedAt: new Date() },
      });
    } catch (error) {
      console.error(`Failed to publish outbox event ${event.id}:`, error);
      // Will be retried on next poll cycle
    }
  }
}

// Run poller on interval
setInterval(processOutbox, 5000); // Every 5 seconds

原因： 双写问题是事件驱动系统中数据不一致最常见的来源。如果先写入数据库，再发布事件，那么即使数据库写入成功，事件发布仍可能失败。发件箱模式可确保只要业务数据已写入，事件最终就会发布。

---

模式 2：幂等事件消费者

适用场景： 适用于所有事件消费者。所有主流消息代理都至少会投递一次消息。

实现：

// Kafka consumer with idempotency
import { Kafka, EachMessagePayload } from "kafkajs";

const kafka = new Kafka({
  clientId: "order-service",
  brokers: [process.env.KAFKA_BROKERS!],
});

const consumer = kafka.consumer({ groupId: "payment-processor" });

interface DomainEvent {
  eventId: string;
  eventType: string;
  aggregateId: string;
  payload: Record<string, unknown>;
  timestamp: string;
}

// Event handler registry
const handlers: Record<string, (event: DomainEvent) => Promise<void>> = {
  OrderPlaced: async (event) => {
    const { orderId, total, customerId } = event.payload as {
      orderId: string;
      total: number;
      customerId: string;
    };

    // Create payment intent (idempotent via orderId)
    await processPayment(orderId, total, customerId);
  },

  PaymentFailed: async (event) => {
    const { orderId } = event.payload as { orderId: string };
    await cancelOrder(orderId);
  },
};

async function startConsumer() {
  await consumer.connect();
  await consumer.subscribe({
    topics: ["Order.OrderPlaced", "Payment.PaymentFailed"],
    fromBeginning: false,
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }: EachMessagePayload) => {
      const event: DomainEvent = JSON.parse(message.value!.toString());

      // 1. Idempotency check - have we already processed this event?
      const alreadyProcessed = await prisma.processedEvent.findUnique({
        where: { eventId: event.eventId },
      });

      if (alreadyProcessed) {
        console.log(`Skipping duplicate event: ${event.eventId}`);
        return;
      }

      // 2. Process the event
      const handler = handlers[event.eventType];
      if (!handler) {
        console.warn(`No handler for event type: ${event.eventType}`);
        return;
      }

      try {
        await prisma.$transaction(async (tx) => {
          // Record event as processed (within the same transaction as business logic)
          await tx.processedEvent.create({
            data: {
              eventId: event.eventId,
              eventType: event.eventType,
              processedAt: new Date(),
            },
          });

          // Execute business logic
          await handler(event);
        });
      } catch (error) {
        console.error(`Error processing event ${event.eventId}:`, error);
        // Don't commit offset - message will be redelivered
        throw error;
      }
    },
  });
}

原因： Kafka、RabbitMQ 和 SQS 都保证至少投递一次，这意味着在重新均衡、重试和网络问题期间，消费者可能会收到重复消息。在同一事务中将已处理的事件 ID 与业务逻辑一并记录，可确保实现恰好一次处理语义。

---

模式 3：Saga 编排

适用场景： 业务流程跨越多个服务，并且需要在任一步骤失败时协调回滚。

实现：

// Saga orchestrator for order fulfillment
// Order -> Payment -> Inventory -> Shipping

interface SagaStep {
  name: string;
  execute: (context: SagaContext) => Promise<void>;
  compensate: (context: SagaContext) => Promise<void>;
}

interface SagaContext {
  orderId: string;
  customerId: string;
  items: OrderItem[];
  total: number;
  // Accumulated state from previous steps
  paymentId?: string;
  reservationId?: string;
  shipmentId?: string;
}

const orderFulfillmentSaga: SagaStep[] = [
  {
    name: "processPayment",
    execute: async (ctx) => {
      const payment = await paymentService.charge({
        orderId: ctx.orderId,
        amount: ctx.total,
        customerId: ctx.customerId,
      });
      ctx.paymentId = payment.id;
    },
    compensate: async (ctx) => {
      if (ctx.paymentId) {
        await paymentService.refund(ctx.paymentId);
      }
    },
  },
  {
    name: "reserveInventory",
    execute: async (ctx) => {
      const reservation = await inventoryService.reserve({
        orderId: ctx.orderId,
        items: ctx.items,
      });
      ctx.reservationId = reservation.id;
    },
    compensate: async (ctx) => {
      if (ctx.reservationId) {
        await inventoryService.release(ctx.reservationId);
      }
    },
  },
  {
    name: "createShipment",
    execute: async (ctx) => {
      const shipment = await shippingService.create({
        orderId: ctx.orderId,
        items: ctx.items,
        customerId: ctx.customerId,
      });
      ctx.shipmentId = shipment.id;
    },
    compensate: async (ctx) => {
      if (ctx.shipmentId) {
        await shippingService.cancel(ctx.shipmentId);
      }
    },
  },
];

async function executeSaga(
  steps: SagaStep[],
  context: SagaContext
): Promise<void> {
  const completedSteps: SagaStep[] = [];

  for (const step of steps) {
    try {
      console.log(`Executing saga step: ${step.name}`);
      await step.execute(context);
      completedSteps.push(step);
    } catch (error) {
      console.error(`Saga step "${step.name}" failed:`, error);

      // Compensate in reverse order
      for (const completedStep of completedSteps.reverse()) {
        try {
          console.log(`Compensating: ${completedStep.name}`);
          await completedStep.compensate(context);
        } catch (compensateError) {
          // Log compensation failure for manual intervention
          console.error(
            `CRITICAL: Compensation for "${completedStep.name}" failed:`,
            compensateError
          );
          await alertOps({
            type: "saga_compensation_failure",
            saga: "orderFulfillment",
            step: completedStep.name,
            orderId: context.orderId,
            error: compensateError,
          });
        }
      }

      throw new SagaFailedError(step.name, error as Error);
    }
  }
}

原因： 跨服务的分布式事务不切实际（两阶段提交无法扩展）。Saga 提供了一种替代方案：由一系列本地事务和用于回滚的补偿操作组成。如果付款成功，但库存预留失败，Saga 会自动退款。

---

模式 4：死信队列处理

适用场景： 每个事件驱动系统都需要为无法处理的消息制定策略。

实现：

// SQS dead letter queue handling pattern
import { SQSClient, SendMessageCommand, ReceiveMessageCommand } from "@aws-sdk/client-sqs";

// Configure main queue with DLQ
const queueConfig = {
  RedrivePolicy: JSON.stringify({
    deadLetterTargetArn: process.env.DLQ_ARN,
    maxReceiveCount: 3, // Move to DLQ after 3 failed processing attempts
  }),
};

// DLQ monitoring and replay
async function monitorDLQ(): Promise<void> {
  const messages = await sqs.send(new ReceiveMessageCommand({
    QueueUrl: process.env.DLQ_URL,
    MaxNumberOfMessages: 10,
    WaitTimeSeconds: 20,
    MessageAttributeNames: ["All"],
  }));

  if (messages.Messages && messages.Messages.length > 0) {
    // Alert operations team
    await alertOps({
      type: "dlq_messages",
      count: messages.Messages.length,
      queue: "order-events-dlq",
      sample: messages.Messages[0].Body,
    });
  }
}

// Replay DLQ messages back to the main queue
async function replayDLQMessages(
  count: number,
  filter?: (msg: Record<string, unknown>) => boolean
): Promise<{ replayed: number; skipped: number }> {
  let replayed = 0;
  let skipped = 0;

  for (let i = 0; i < count; i++) {
    const response = await sqs.send(new ReceiveMessageCommand({
      QueueUrl: process.env.DLQ_URL,
      MaxNumberOfMessages: 1,
      WaitTimeSeconds: 0,
    }));

    const message = response.Messages?.[0];
    if (!message) break;

    const body = JSON.parse(message.Body!);

    if (filter && !filter(body)) {
      skipped++;
      continue;
    }

    // Replay to main queue
    await sqs.send(new SendMessageCommand({
      QueueUrl: process.env.MAIN_QUEUE_URL,
      MessageBody: message.Body!,
      MessageAttributes: message.MessageAttributes,
    }));

    // Delete from DLQ
    await sqs.send(new DeleteMessageCommand({
      QueueUrl: process.env.DLQ_URL,
      ReceiptHandle: message.ReceiptHandle!,
    }));

    replayed++;
  }

  return { replayed, skipped };
}

原因： 死信队列会捕获所有重试后仍处理失败的消息。没有 DLQ，失败的消息会阻塞队列或被静默丢弃。监控 DLQ 有助于发现程序错误（格式错误的事件、缺失的处理程序）。重放工具可在不丢失数据的情况下，从暂时性问题（下游服务故障）中恢复。

---

消息代理选择指南

消息代理	最适用场景	顺序保证	吞吐量	持久性
Kafka	高吞吐量事件流、日志压缩	按分区	非常高（每秒数百万条）	保留策略可配置
RabbitMQ	任务队列、路由、请求-响应	按队列	高（每秒 10 万条）	可选
SQS/SNS	无服务器、原生 AWS、低运维开销	按 FIFO 队列	高（FIFO 每秒 3,000 条，标准队列无限制）	保留 14 天
NATS	低延迟、云原生、轻量级	JetStream	非常高	通过 JetStream 实现持久性

---

反模式

反模式	不良影响	更好的做法
没有使用 outbox 的双重写入（数据库 + 事件）	任一写入失败都会导致数据不一致	事务性 outbox 模式
消费者没有实现幂等性	重新投递时会重复处理	在同一事务中按事件 ID 去重
事件负载过大（> 1 MB）	处理缓慢，受代理限制	将数据存储在服务中，发送引用（索取检查模式）
同步处理事件	背离异步架构的初衷	使用队列异步处理事件
未配置 DLQ	失败的消息会阻塞队列或消失	始终配置 DLQ 并设置告警
事件架构紧密耦合	架构变更会破坏消费者	以向后兼容的方式演进，并显式进行版本控制
从多个位置发布领域事件	事件格式不一致，或遗漏事件	统一发布点（聚合根 / 服务层）

---

检查清单

• 使用事务性 outbox 模式，原子化地写入数据库和事件
• 所有消费者均实现幂等性（按事件 ID 去重）
• 配置死信队列、告警和重放工具
• 对事件架构进行版本控制，并以向后兼容的方式演进
• 为每个服务配置消费者组
• 在进入 DLQ 前采用带指数退避的重试策略
• 使用 Saga 或编排模式处理跨服务工作流
• 监控：消费者滞后、DLQ 深度、处理错误
• 在需要时保留事件顺序（分区键）
• 消息代理集群具备容错能力（复制）

---

相关资源

• 技能： monitoring-observability（消费者滞后监控）、serverless-development（SQS/Lambda）
• 技能： payment-integration（支付事件流）、email-systems（将电子邮件作为异步事件）
• 规则： docs/reference/stacks/fullstack-nextjs-nestjs.md（NestJS CQRS 模块）