1. 事件驱动架构基础

事件驱动架构（EDA）是一种软件设计范式，系统通过产生和消费事件来进行通信。事件代表状态的变化——"已经发生的事实"。与请求-响应模型不同，EDA 中的生产者不知道也不关心谁消费了事件。

核心概念

// Event interface — the foundation of EDA
interface DomainEvent {
  eventId: string;           // Unique event identifier (UUID)
  eventType: string;         // e.g., "OrderPlaced"
  aggregateId: string;       // ID of the entity that produced the event
  aggregateType: string;     // e.g., "Order"
  timestamp: string;         // ISO 8601 timestamp
  version: number;           // Schema version for evolution
  payload: Record<string, unknown>;
  metadata: {
    correlationId: string;   // Traces a request across services
    causationId: string;     // ID of the event/command that caused this
    userId?: string;
  };
}

// Example: OrderPlaced event
const orderPlaced: DomainEvent = {
  eventId: "evt-a1b2c3d4",
  eventType: "OrderPlaced",
  aggregateId: "order-12345",
  aggregateType: "Order",
  timestamp: "2026-02-28T10:30:00Z",
  version: 1,
  payload: {
    customerId: "cust-67890",
    items: [
      { productId: "prod-001", quantity: 2, price: 29.99 },
      { productId: "prod-002", quantity: 1, price: 49.99 },
    ],
    totalAmount: 109.97,
    currency: "USD",
  },
  metadata: {
    correlationId: "corr-xyz-789",
    causationId: "cmd-place-order-456",
    userId: "user-abc",
  },
};

2. Apache Kafka 深入剖析

Apache Kafka 是一个分布式事件流平台，设计用于高吞吐量、持久化和可重放的事件处理。Kafka 将数据组织为主题（Topics），每个主题分为多个分区（Partitions），实现并行处理和水平扩展。

主题、分区和消费者组

// Kafka producer — Node.js with kafkajs
import { Kafka, Partitioners } from "kafkajs";

const kafka = new Kafka({
  clientId: "order-service",
  brokers: ["kafka-1:9092", "kafka-2:9092", "kafka-3:9092"],
});

const producer = kafka.producer({
  createPartitioner: Partitioners.DefaultPartitioner,
  idempotent: true,            // Enable exactly-once semantics
  maxInFlightRequests: 5,
  transactionalId: "order-producer-txn",
});

async function publishOrderEvent(order: Order): Promise<void> {
  await producer.connect();

  // Key determines partition — same orderId always goes to same partition
  await producer.send({
    topic: "orders.events",
    messages: [
      {
        key: order.id,
        value: JSON.stringify({
          eventType: "OrderPlaced",
          payload: order,
          timestamp: new Date().toISOString(),
        }),
        headers: {
          "correlation-id": order.correlationId,
          "event-type": "OrderPlaced",
          "schema-version": "1",
        },
      },
    ],
  });
}

// Kafka consumer with consumer group
const consumer = kafka.consumer({
  groupId: "inventory-service-group",
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
  maxBytesPerPartition: 1048576, // 1MB
});

async function startConsumer(): Promise<void> {
  await consumer.connect();
  await consumer.subscribe({
    topics: ["orders.events"],
    fromBeginning: false,
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const event = JSON.parse(message.value!.toString());
      const correlationId = message.headers?.["correlation-id"]?.toString();

      console.log(
        `[Partition \${partition}] Processing \${event.eventType}`,
        `offset=\${message.offset}, key=\${message.key}`
      );

      switch (event.eventType) {
        case "OrderPlaced":
          await reserveInventory(event.payload);
          break;
        case "OrderCancelled":
          await releaseInventory(event.payload);
          break;
        default:
          console.warn(`Unknown event type: \${event.eventType}`);
      }
    },
  });
}

Kafka 精确一次语义

Kafka 通过幂等生产者和事务 API 实现精确一次语义（EOS），确保消息在生产者重试时不会重复写入，并且消费者可以在事务提交后才看到消息。

// Exactly-once: transactional produce + consume
async function processAndForward(inputTopic: string, outputTopic: string) {
  const transaction = await producer.transaction();

  try {
    // Read from input topic
    const messages = await consumeBatch(inputTopic);

    for (const msg of messages) {
      const enriched = await enrichEvent(msg);

      // Write to output topic within the same transaction
      await transaction.send({
        topic: outputTopic,
        messages: [{ key: msg.key, value: JSON.stringify(enriched) }],
      });
    }

    // Commit offsets and produced messages atomically
    await transaction.sendOffsets({
      consumerGroupId: "enrichment-group",
      topics: [{ topic: inputTopic, partitions: messages.map(m => ({
        partition: m.partition,
        offset: (parseInt(m.offset) + 1).toString(),
      }))}],
    });

    await transaction.commit();
  } catch (err) {
    await transaction.abort();
    throw err;
  }
}

3. RabbitMQ 消息模式

RabbitMQ 是一个功能丰富的消息代理，通过交换器（Exchange）和队列（Queue）提供灵活的路由模式。它支持多种交换器类型，每种适用于不同的消息路由场景。

// RabbitMQ with amqplib — Topic exchange + Dead Letter Queue
import amqp from "amqplib";

async function setupRabbitMQ() {
  const connection = await amqp.connect("amqp://localhost:5672");
  const channel = await connection.createChannel();

  // Dead letter exchange for failed messages
  await channel.assertExchange("dlx.events", "fanout", { durable: true });
  await channel.assertQueue("dlq.events", {
    durable: true,
    arguments: { "x-message-ttl": 86400000 }, // 24h retention
  });
  await channel.bindQueue("dlq.events", "dlx.events", "");

  // Main topic exchange
  await channel.assertExchange("domain.events", "topic", { durable: true });

  // Order events queue with DLQ
  await channel.assertQueue("order.processing", {
    durable: true,
    arguments: {
      "x-dead-letter-exchange": "dlx.events",
      "x-dead-letter-routing-key": "order.failed",
      "x-max-retries": 3,
    },
  });

  // Bind queue with topic pattern
  // order.* matches order.placed, order.shipped, etc.
  await channel.bindQueue("order.processing", "domain.events", "order.*");

  // Publish an event
  channel.publish(
    "domain.events",
    "order.placed",
    Buffer.from(JSON.stringify({
      eventType: "OrderPlaced",
      orderId: "order-12345",
      timestamp: new Date().toISOString(),
    })),
    {
      persistent: true,
      messageId: "msg-uuid-123",
      contentType: "application/json",
      headers: { "x-retry-count": 0 },
    }
  );

  // Consumer with manual acknowledgment
  await channel.prefetch(10); // Process 10 messages at a time
  await channel.consume("order.processing", async (msg) => {
    if (!msg) return;
    try {
      const event = JSON.parse(msg.content.toString());
      await processOrderEvent(event);
      channel.ack(msg); // Acknowledge success
    } catch (err) {
      const retryCount = (msg.properties.headers?.["x-retry-count"] || 0) + 1;
      if (retryCount >= 3) {
        channel.nack(msg, false, false); // Send to DLQ
      } else {
        channel.nack(msg, false, true);   // Requeue for retry
      }
    }
  });
}

4. 事件溯源

事件溯源将状态存储为一系列不可变事件，而非仅保存当前状态。当前状态通过按顺序重放事件来推导。这提供了完整的审计日志、时间旅行查询能力，以及从事件流构建多个读取优化视图的能力。

事件存储实现

// Event Store — append-only with optimistic concurrency
interface StoredEvent {
  streamId: string;
  version: number;
  eventType: string;
  payload: Record<string, unknown>;
  metadata: Record<string, string>;
  timestamp: Date;
}

class EventStore {
  constructor(private db: Database) {}

  async appendToStream(
    streamId: string,
    events: StoredEvent[],
    expectedVersion: number
  ): Promise<void> {
    const currentVersion = await this.getStreamVersion(streamId);

    // Optimistic concurrency check
    if (currentVersion !== expectedVersion) {
      throw new ConcurrencyError(
        `Expected version \${expectedVersion}, got \${currentVersion}`
      );
    }

    const transaction = await this.db.beginTransaction();
    try {
      for (let i = 0; i < events.length; i++) {
        await transaction.execute(
          `INSERT INTO events (stream_id, version, event_type, payload, metadata, timestamp)
           VALUES (?, ?, ?, ?, ?, ?)`,
          [
            streamId,
            expectedVersion + i + 1,
            events[i].eventType,
            JSON.stringify(events[i].payload),
            JSON.stringify(events[i].metadata),
            events[i].timestamp,
          ]
        );
      }
      await transaction.commit();
    } catch (err) {
      await transaction.rollback();
      throw err;
    }
  }

  async readStream(streamId: string, fromVersion = 0): Promise<StoredEvent[]> {
    return this.db.query(
      `SELECT * FROM events
       WHERE stream_id = ? AND version > ?
       ORDER BY version ASC`,
      [streamId, fromVersion]
    );
  }
}

投影和快照

// Aggregate with snapshot support
class OrderAggregate {
  private state: OrderState = { status: "draft", items: [], total: 0 };
  private version = 0;
  private uncommittedEvents: StoredEvent[] = [];

  // Rebuild from events with snapshot optimization
  static async load(
    eventStore: EventStore,
    snapshotStore: SnapshotStore,
    orderId: string
  ): Promise<OrderAggregate> {
    const aggregate = new OrderAggregate();

    // Try loading from snapshot first
    const snapshot = await snapshotStore.get(orderId);
    if (snapshot) {
      aggregate.state = snapshot.state;
      aggregate.version = snapshot.version;
    }

    // Replay events after the snapshot
    const events = await eventStore.readStream(orderId, aggregate.version);
    for (const event of events) {
      aggregate.apply(event, false);
    }

    // Take a new snapshot every 100 events
    if (events.length > 100) {
      await snapshotStore.save(orderId, aggregate.state, aggregate.version);
    }

    return aggregate;
  }

  placeOrder(customerId: string, items: OrderItem[]): void {
    if (this.state.status !== "draft") {
      throw new Error("Order already placed");
    }
    this.apply({
      eventType: "OrderPlaced",
      payload: { customerId, items, total: this.calculateTotal(items) },
    } as StoredEvent, true);
  }

  private apply(event: StoredEvent, isNew: boolean): void {
    // State mutation based on event type
    switch (event.eventType) {
      case "OrderPlaced":
        this.state.status = "placed";
        this.state.items = event.payload.items as OrderItem[];
        this.state.total = event.payload.total as number;
        break;
      case "OrderShipped":
        this.state.status = "shipped";
        break;
      case "OrderCancelled":
        this.state.status = "cancelled";
        break;
    }
    this.version++;
    if (isNew) this.uncommittedEvents.push(event);
  }
}

投影（读取模型）

// Projection — build read-optimized views from events
class OrderSummaryProjection {
  constructor(private readDb: Database) {}

  async handle(event: StoredEvent): Promise<void> {
    switch (event.eventType) {
      case "OrderPlaced":
        await this.readDb.execute(
          `INSERT INTO order_summaries
           (order_id, customer_id, status, total, item_count, placed_at)
           VALUES (?, ?, ?, ?, ?, ?)`,
          [
            event.streamId,
            event.payload.customerId,
            "placed",
            event.payload.total,
            (event.payload.items as unknown[]).length,
            event.timestamp,
          ]
        );
        break;

      case "OrderShipped":
        await this.readDb.execute(
          `UPDATE order_summaries
           SET status = ?, shipped_at = ?
           WHERE order_id = ?`,
          ["shipped", event.timestamp, event.streamId]
        );
        break;

      case "OrderCancelled":
        await this.readDb.execute(
          `UPDATE order_summaries
           SET status = ?, cancelled_at = ?
           WHERE order_id = ?`,
          ["cancelled", event.timestamp, event.streamId]
        );
        break;
    }
  }
}

5. CQRS 实现

CQRS（命令查询职责分离）将写操作（命令）和读操作（查询）分为不同的模型。写模型优化数据一致性和业务规则验证，读模型优化查询性能。两个模型通过事件进行同步。

// CQRS — Command side
interface Command {
  type: string;
  payload: Record<string, unknown>;
  metadata: { userId: string; correlationId: string };
}

class CommandBus {
  private handlers = new Map<string, CommandHandler>();

  register(commandType: string, handler: CommandHandler): void {
    this.handlers.set(commandType, handler);
  }

  async dispatch(command: Command): Promise<void> {
    const handler = this.handlers.get(command.type);
    if (!handler) throw new Error(`No handler for: \${command.type}`);
    await handler.execute(command);
  }
}

class PlaceOrderHandler implements CommandHandler {
  constructor(
    private eventStore: EventStore,
    private snapshotStore: SnapshotStore
  ) {}

  async execute(command: Command): Promise<void> {
    const { customerId, items } = command.payload;

    // Load aggregate from event store
    const order = await OrderAggregate.load(
      this.eventStore, this.snapshotStore, command.payload.orderId as string
    );

    // Execute business logic (may throw if invalid)
    order.placeOrder(customerId as string, items as OrderItem[]);

    // Persist new events
    await this.eventStore.appendToStream(
      command.payload.orderId as string,
      order.getUncommittedEvents(),
      order.getVersion()
    );
  }
}

// CQRS — Query side
interface Query {
  type: string;
  params: Record<string, unknown>;
}

class QueryBus {
  private handlers = new Map<string, QueryHandler>();

  register(queryType: string, handler: QueryHandler): void {
    this.handlers.set(queryType, handler);
  }

  async dispatch<T>(query: Query): Promise<T> {
    const handler = this.handlers.get(query.type);
    if (!handler) throw new Error(`No handler for: \${query.type}`);
    return handler.execute(query) as Promise<T>;
  }
}

class GetOrderSummaryHandler implements QueryHandler {
  constructor(private readDb: Database) {}

  async execute(query: Query): Promise<OrderSummary> {
    // Query the read-optimized projection table
    const result = await this.readDb.queryOne(
      `SELECT order_id, customer_id, status, total, item_count,
              placed_at, shipped_at, cancelled_at
       FROM order_summaries WHERE order_id = ?`,
      [query.params.orderId]
    );
    if (!result) throw new NotFoundError("Order not found");
    return result;
  }
}

6. Saga 模式

Saga 模式用于管理跨多个服务的分布式事务。每个 Saga 步骤是一个本地事务，配有对应的补偿操作用于回滚。Saga 有两种实现方式：编排式（中央协调器）和协调式（事件驱动）。

编排式 Saga 实现

// Orchestration Saga — Order fulfillment
type SagaStep = {
  name: string;
  execute: (context: SagaContext) => Promise<void>;
  compensate: (context: SagaContext) => Promise<void>;
};

class OrderSaga {
  private steps: SagaStep[] = [
    {
      name: "ReserveInventory",
      execute: async (ctx) => {
        ctx.inventoryReservationId = await inventoryService.reserve(
          ctx.orderId, ctx.items
        );
      },
      compensate: async (ctx) => {
        await inventoryService.cancelReservation(ctx.inventoryReservationId);
      },
    },
    {
      name: "ProcessPayment",
      execute: async (ctx) => {
        ctx.paymentId = await paymentService.charge(
          ctx.customerId, ctx.totalAmount
        );
      },
      compensate: async (ctx) => {
        await paymentService.refund(ctx.paymentId);
      },
    },
    {
      name: "ArrangeShipping",
      execute: async (ctx) => {
        ctx.shipmentId = await shippingService.createShipment(
          ctx.orderId, ctx.shippingAddress
        );
      },
      compensate: async (ctx) => {
        await shippingService.cancelShipment(ctx.shipmentId);
      },
    },
  ];

  async run(context: SagaContext): Promise<void> {
    const completedSteps: SagaStep[] = [];

    for (const step of this.steps) {
      try {
        console.log(`Executing step: \${step.name}`);
        await step.execute(context);
        completedSteps.push(step);
      } catch (error) {
        console.error(`Step \${step.name} failed:`, error);
        // Compensate in reverse order
        for (const completed of completedSteps.reverse()) {
          try {
            console.log(`Compensating step: \${completed.name}`);
            await completed.compensate(context);
          } catch (compError) {
            console.error(`Compensation failed for \${completed.name}:`, compError);
            // Log to dead letter queue for manual intervention
            await deadLetterQueue.publish({
              sagaId: context.sagaId,
              failedStep: completed.name,
              error: compError,
            });
          }
        }
        throw new SagaFailedError(step.name, error);
      }
    }
  }
}

7. DDD 中的领域事件

在领域驱动设计（DDD）中，领域事件代表业务领域中有意义的状态变化。聚合根在执行命令后产生领域事件，事件处理器在同一限界上下文或跨限界上下文中响应这些事件，实现最终一致性。

// Domain Events in DDD — Aggregate producing events
abstract class AggregateRoot {
  private domainEvents: DomainEvent[] = [];
  protected version = 0;

  protected addDomainEvent(event: DomainEvent): void {
    this.domainEvents.push(event);
  }

  getDomainEvents(): DomainEvent[] {
    return [...this.domainEvents];
  }

  clearDomainEvents(): void {
    this.domainEvents = [];
  }
}

class Order extends AggregateRoot {
  private status: OrderStatus = "draft";
  private items: OrderItem[] = [];

  place(customerId: string, items: OrderItem[]): void {
    // Business rule validation
    if (items.length === 0) throw new Error("Order must have items");
    if (this.status !== "draft") throw new Error("Order already placed");

    this.status = "placed";
    this.items = items;

    this.addDomainEvent({
      eventId: generateUUID(),
      eventType: "OrderPlaced",
      aggregateId: this.id,
      aggregateType: "Order",
      timestamp: new Date().toISOString(),
      version: ++this.version,
      payload: { customerId, items, total: this.calculateTotal() },
      metadata: { correlationId: generateUUID(), causationId: "" },
    });
  }

  cancel(reason: string): void {
    if (this.status === "shipped") {
      throw new Error("Cannot cancel shipped order");
    }
    this.status = "cancelled";

    this.addDomainEvent({
      eventId: generateUUID(),
      eventType: "OrderCancelled",
      aggregateId: this.id,
      aggregateType: "Order",
      timestamp: new Date().toISOString(),
      version: ++this.version,
      payload: { reason },
      metadata: { correlationId: generateUUID(), causationId: "" },
    });
  }
}

// Event dispatcher — publish domain events after persistence
class DomainEventDispatcher {
  private handlers = new Map<string, DomainEventHandler[]>();

  subscribe(eventType: string, handler: DomainEventHandler): void {
    const existing = this.handlers.get(eventType) || [];
    this.handlers.set(eventType, [...existing, handler]);
  }

  async dispatch(events: DomainEvent[]): Promise<void> {
    for (const event of events) {
      const handlers = this.handlers.get(event.eventType) || [];
      await Promise.all(
        handlers.map(h => h.handle(event))
      );
    }
  }
}

8. 异步消息模式

异步消息模式定义了服务之间交换消息的方式。选择正确的模式取决于消息的语义（事件 vs 命令）、消费者数量和交付保证要求。

// Request-Reply pattern with correlation IDs
class AsyncRequestReply {
  private pendingRequests = new Map<string, {
    resolve: (value: unknown) => void;
    reject: (reason: unknown) => void;
    timer: NodeJS.Timeout;
  }>();

  constructor(
    private channel: MessageChannel,
    private replyQueue: string,
    private timeoutMs = 30000
  ) {
    // Listen for replies on the dedicated reply queue
    this.channel.consume(this.replyQueue, (msg) => {
      if (!msg) return;
      const correlationId = msg.properties.correlationId;
      const pending = this.pendingRequests.get(correlationId);

      if (pending) {
        clearTimeout(pending.timer);
        this.pendingRequests.delete(correlationId);
        const reply = JSON.parse(msg.content.toString());
        if (reply.error) {
          pending.reject(new Error(reply.error));
        } else {
          pending.resolve(reply.data);
        }
      }
      this.channel.ack(msg);
    });
  }

  async request(queue: string, payload: unknown): Promise<unknown> {
    const correlationId = generateUUID();

    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.pendingRequests.delete(correlationId);
        reject(new Error("Request timed out"));
      }, this.timeoutMs);

      this.pendingRequests.set(correlationId, { resolve, reject, timer });

      this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(payload)), {
        correlationId,
        replyTo: this.replyQueue,
        expiration: String(this.timeoutMs),
      });
    });
  }
}

9. 事件 Schema 演进

在长期运行的事件驱动系统中，事件 Schema 不可避免地需要演进。Schema 演进的关键是维护向后兼容性，确保新消费者能读取旧事件，旧消费者不会被新事件破坏。

// Avro schema evolution example
// Version 1 — Original OrderPlaced schema
const orderPlacedV1 = {
  type: "record",
  name: "OrderPlaced",
  namespace: "com.example.events",
  fields: [
    { name: "orderId", type: "string" },
    { name: "customerId", type: "string" },
    { name: "totalAmount", type: "double" },
    { name: "currency", type: "string" },
    { name: "timestamp", type: "string" },
  ],
};

// Version 2 — Added optional fields (backward compatible)
const orderPlacedV2 = {
  type: "record",
  name: "OrderPlaced",
  namespace: "com.example.events",
  fields: [
    { name: "orderId", type: "string" },
    { name: "customerId", type: "string" },
    { name: "totalAmount", type: "double" },
    { name: "currency", type: "string" },
    { name: "timestamp", type: "string" },
    // New optional fields with defaults — backward compatible
    { name: "discountCode", type: ["null", "string"], default: null },
    { name: "priority", type: "string", default: "normal" },
    { name: "channel", type: "string", default: "web" },
  ],
};

// Schema Registry client
import { SchemaRegistry } from "@kafkajs/confluent-schema-registry";

const registry = new SchemaRegistry({
  host: "http://schema-registry:8081",
});

// Register schema with compatibility check
async function registerSchema() {
  const { id } = await registry.register({
    type: "AVRO",
    schema: JSON.stringify(orderPlacedV2),
  }, {
    subject: "orders.events-value",
  });
  console.log("Registered schema with id:", id);
}

// Produce with schema validation
async function produceWithSchema(event: OrderPlacedEvent) {
  const schemaId = await registry.getLatestSchemaId("orders.events-value");
  const encodedValue = await registry.encode(schemaId, event);

  await producer.send({
    topic: "orders.events",
    messages: [{ key: event.orderId, value: encodedValue }],
  });
}

10. 无服务器事件处理

无服务器平台天然适合事件驱动架构，因为函数本身就是事件处理器。AWS Lambda + EventBridge 和 Azure Functions + Event Grid 是两种主流的无服务器事件处理方案。

// AWS Lambda + EventBridge — Order event processing
// serverless.yml (Serverless Framework)
// service: order-processor
// provider:
//   name: aws
//   runtime: nodejs20.x
// functions:
//   processOrder:
//     handler: handler.processOrder
//     events:
//       - eventBridge:
//           pattern:
//             source: ["order-service"]
//             detail-type: ["OrderPlaced", "OrderCancelled"]

import { EventBridgeEvent, Context } from "aws-lambda";
import { EventBridgeClient, PutEventsCommand } from "@aws-sdk/client-eventbridge";

const eventBridge = new EventBridgeClient({});

// Producer — publish event to EventBridge
export async function publishOrderEvent(order: Order): Promise<void> {
  await eventBridge.send(new PutEventsCommand({
    Entries: [{
      Source: "order-service",
      DetailType: "OrderPlaced",
      Detail: JSON.stringify({
        orderId: order.id,
        customerId: order.customerId,
        totalAmount: order.total,
        items: order.items,
      }),
      EventBusName: "orders-bus",
    }],
  }));
}

// Consumer — Lambda handler triggered by EventBridge
export async function processOrder(
  event: EventBridgeEvent<"OrderPlaced", OrderPayload>,
  context: Context
): Promise<void> {
  const { orderId, customerId, totalAmount } = event.detail;

  console.log(
    `Processing \${event["detail-type"]} for order \${orderId}`,
    `requestId=\${context.awsRequestId}`
  );

  // Idempotency check — prevent duplicate processing
  const alreadyProcessed = await checkIdempotencyKey(orderId);
  if (alreadyProcessed) {
    console.log(`Order \${orderId} already processed, skipping`);
    return;
  }

  await reserveInventory(orderId);
  await sendConfirmationEmail(customerId, orderId);
  await storeIdempotencyKey(orderId);

  // Chain next event
  await eventBridge.send(new PutEventsCommand({
    Entries: [{
      Source: "fulfillment-service",
      DetailType: "InventoryReserved",
      Detail: JSON.stringify({ orderId, customerId }),
      EventBusName: "orders-bus",
    }],
  }));
}

11. 流处理

流处理引擎（如 Kafka Streams 和 Apache Flink）可以对无限事件流进行实时的有状态计算，包括窗口聚合、流-流连接和复杂事件处理。

Kafka Streams 示例

// Kafka Streams topology — Java
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.time.Duration;

public class OrderAnalyticsTopology {

  public static Topology build() {
    StreamsBuilder builder = new StreamsBuilder();

    // Source: read order events
    KStream<String, OrderEvent> orders = builder
      .stream("orders.events",
        Consumed.with(Serdes.String(), orderEventSerde));

    // Windowed aggregation: order count per customer per hour
    KTable<Windowed<String>, Long> hourlyOrderCounts = orders
      .filter((key, event) -> "OrderPlaced".equals(event.getType()))
      .selectKey((key, event) -> event.getCustomerId())
      .groupByKey()
      .windowedBy(
        TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1))
      )
      .count(Materialized.as("hourly-order-counts"));

    // Real-time revenue per product
    KTable<String, Double> revenueByProduct = orders
      .filter((key, event) -> "OrderPlaced".equals(event.getType()))
      .flatMapValues(event -> event.getItems())
      .selectKey((key, item) -> item.getProductId())
      .groupByKey()
      .aggregate(
        () -> 0.0,
        (productId, item, total) -> total + item.getPrice() * item.getQuantity(),
        Materialized.as("revenue-by-product")
      );

    // Stream-stream join: match orders with payments within 5 min
    KStream<String, PaymentEvent> payments = builder
      .stream("payments.events",
        Consumed.with(Serdes.String(), paymentEventSerde));

    KStream<String, OrderWithPayment> matched = orders
      .join(
        payments,
        (order, payment) -> new OrderWithPayment(order, payment),
        JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
        StreamJoined.with(Serdes.String(), orderEventSerde, paymentEventSerde)
      );

    matched.to("orders.matched",
      Produced.with(Serdes.String(), orderWithPaymentSerde));

    return builder.build();
  }
}

Apache Flink 窗口示例

// Apache Flink — Python (PyFlink) windowed aggregation
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common.time import Time
from pyflink.common.watermark_strategy import WatermarkStrategy
import json

env = StreamExecutionEnvironment.get_execution_environment()

# Configure Kafka source
kafka_source = KafkaSource.builder() \
    .set_bootstrap_servers("kafka:9092") \
    .set_topics("orders.events") \
    .set_group_id("flink-analytics") \
    .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

orders = env.from_source(
    kafka_source,
    WatermarkStrategy.for_bounded_out_of_orderness(Time.seconds(5)),
    "OrderEvents"
)

# Tumbling window: revenue per minute
revenue_per_minute = orders \
    .map(lambda raw: json.loads(raw)) \
    .filter(lambda e: e["eventType"] == "OrderPlaced") \
    .key_by(lambda e: "global") \
    .window(TumblingEventTimeWindows.of(Time.minutes(1))) \
    .reduce(lambda a, b: {
        "totalRevenue": a["totalRevenue"] + b["payload"]["totalAmount"],
        "orderCount": a["orderCount"] + 1,
    })

revenue_per_minute.print()
env.execute("Order Revenue Analytics")

12. 测试事件驱动系统

测试事件驱动系统需要特殊策略：合约测试验证生产者和消费者之间的 Schema 兼容性，事件重放测试验证投影的正确性，混沌工程测试系统在代理故障时的弹性。

// Contract testing with embedded Kafka (Jest)
import { EmbeddedKafka } from "@testcontainers/kafka";

describe("OrderPlaced event contract", () => {
  let kafka: EmbeddedKafka;

  beforeAll(async () => {
    kafka = await new EmbeddedKafka().start();
  });

  afterAll(async () => {
    await kafka.stop();
  });

  test("producer publishes valid OrderPlaced event", async () => {
    const producer = kafka.producer();
    await producer.connect();

    // Produce event
    await producer.send({
      topic: "orders.events",
      messages: [{
        key: "order-123",
        value: JSON.stringify({
          eventType: "OrderPlaced",
          orderId: "order-123",
          customerId: "cust-456",
          totalAmount: 99.99,
          items: [{ productId: "p1", quantity: 2, price: 49.995 }],
          timestamp: new Date().toISOString(),
        }),
      }],
    });

    // Consume and validate schema
    const consumer = kafka.consumer({ groupId: "test-group" });
    await consumer.connect();
    await consumer.subscribe({ topics: ["orders.events"] });

    const received = await consumeOneMessage(consumer);
    const event = JSON.parse(received.value!.toString());

    // Contract assertions
    expect(event).toHaveProperty("eventType", "OrderPlaced");
    expect(event).toHaveProperty("orderId");
    expect(event).toHaveProperty("customerId");
    expect(event).toHaveProperty("totalAmount");
    expect(typeof event.totalAmount).toBe("number");
    expect(Array.isArray(event.items)).toBe(true);
    expect(event.items[0]).toHaveProperty("productId");
    expect(event.items[0]).toHaveProperty("quantity");
    expect(event.items[0]).toHaveProperty("price");
  });
});

// Event replay testing — verify projections
describe("OrderSummaryProjection", () => {
  test("rebuilds correct state from event sequence", async () => {
    const events: StoredEvent[] = [
      {
        streamId: "order-001",
        version: 1,
        eventType: "OrderPlaced",
        payload: {
          customerId: "cust-1",
          items: [{ productId: "p1", quantity: 2, price: 25 }],
          total: 50,
        },
        metadata: {},
        timestamp: new Date("2026-01-01"),
      },
      {
        streamId: "order-001",
        version: 2,
        eventType: "OrderShipped",
        payload: { trackingNumber: "TRACK-123" },
        metadata: {},
        timestamp: new Date("2026-01-02"),
      },
    ];

    const projection = new OrderSummaryProjection(testDb);
    for (const event of events) {
      await projection.handle(event);
    }

    const summary = await testDb.queryOne(
      "SELECT * FROM order_summaries WHERE order_id = ?",
      ["order-001"]
    );

    expect(summary.status).toBe("shipped");
    expect(summary.total).toBe(50);
    expect(summary.item_count).toBe(1);
  });
});

混沌工程测试

// Chaos engineering — simulate broker failures
describe("Resilience under Kafka broker failure", () => {
  test("consumer recovers after broker restart", async () => {
    const kafka = await new KafkaContainer()
      .withExposedPorts(9092)
      .start();

    // Produce messages
    await produceTestMessages(kafka, 100);

    // Start consuming
    const consumed: string[] = [];
    const consumer = startTestConsumer(kafka, (msg) => {
      consumed.push(msg);
    });

    // Wait for partial consumption
    await waitUntil(() => consumed.length >= 50);

    // Simulate broker crash
    await kafka.stop();
    await delay(5000); // Broker is down for 5 seconds
    await kafka.start();

    // Verify consumer reconnects and processes remaining messages
    await waitUntil(() => consumed.length === 100, 30000);
    expect(consumed.length).toBe(100);

    // Verify no duplicates (idempotent processing)
    const uniqueIds = new Set(consumed.map(m => JSON.parse(m).eventId));
    expect(uniqueIds.size).toBe(100);
  });
});

13. 监控与可观测性

事件驱动系统的可观测性围绕三个关键维度：分布式追踪（跨服务的请求链路）、消费者延迟监控（消息处理速度）、以及死信队列警报（失败消息）。

// Distributed tracing with OpenTelemetry
import { trace, context, propagation, SpanKind } from "@opentelemetry/api";
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { JaegerExporter } from "@opentelemetry/exporter-jaeger";

const tracer = trace.getTracer("order-service", "1.0.0");

// Producer — inject trace context into message headers
async function produceWithTracing(
  topic: string,
  event: DomainEvent
): Promise<void> {
  const span = tracer.startSpan(`publish \${event.eventType}`, {
    kind: SpanKind.PRODUCER,
    attributes: {
      "messaging.system": "kafka",
      "messaging.destination": topic,
      "messaging.message_id": event.eventId,
      "event.type": event.eventType,
    },
  });

  // Inject trace context into Kafka headers
  const headers: Record<string, string> = {};
  propagation.inject(context.active(), headers);

  try {
    await producer.send({
      topic,
      messages: [{
        key: event.aggregateId,
        value: JSON.stringify(event),
        headers,
      }],
    });
    span.setStatus({ code: 0 }); // OK
  } catch (err) {
    span.setStatus({ code: 2, message: String(err) }); // ERROR
    throw err;
  } finally {
    span.end();
  }
}

// Consumer — extract trace context from message headers
async function consumeWithTracing(
  message: KafkaMessage
): Promise<void> {
  // Extract parent trace context from Kafka headers
  const parentCtx = propagation.extract(
    context.active(),
    message.headers
  );

  const span = tracer.startSpan(
    "process OrderPlaced",
    { kind: SpanKind.CONSUMER },
    parentCtx
  );

  return context.with(
    trace.setSpan(parentCtx, span),
    async () => {
      try {
        const event = JSON.parse(message.value!.toString());
        await processEvent(event);
        span.setStatus({ code: 0 });
      } catch (err) {
        span.setStatus({ code: 2, message: String(err) });
        throw err;
      } finally {
        span.end();
      }
    }
  );
}

消费者延迟和死信队列监控

// Consumer lag monitoring with Prometheus metrics
import { Counter, Gauge, Histogram } from "prom-client";

const consumerLag = new Gauge({
  name: "kafka_consumer_lag",
  help: "Consumer lag in messages",
  labelNames: ["topic", "partition", "consumer_group"],
});

const messagesProcessed = new Counter({
  name: "messages_processed_total",
  help: "Total messages processed",
  labelNames: ["topic", "event_type", "status"],
});

const processingDuration = new Histogram({
  name: "message_processing_duration_seconds",
  help: "Time to process a single message",
  labelNames: ["topic", "event_type"],
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 5, 10],
});

const dlqMessages = new Counter({
  name: "dlq_messages_total",
  help: "Messages sent to dead letter queue",
  labelNames: ["topic", "error_type"],
});

// Monitor consumer lag periodically
async function monitorLag(admin: KafkaAdmin, groupId: string) {
  const offsets = await admin.fetchOffsets({ groupId });
  const topicOffsets = await admin.fetchTopicOffsets("orders.events");

  for (const partition of offsets) {
    const latest = topicOffsets.find(
      t => t.partition === partition.partition
    );
    if (latest) {
      const lag = parseInt(latest.offset) - parseInt(partition.offset);
      consumerLag.set(
        { topic: "orders.events", partition: String(partition.partition), consumer_group: groupId },
        lag
      );
    }
  }
}

// Alerting rules (Prometheus/Grafana)
// ALERT ConsumerLagHigh
//   IF kafka_consumer_lag > 10000
//   FOR 5m
//   LABELS { severity = "warning" }
//   ANNOTATIONS { summary = "Consumer lag exceeding 10K messages" }
//
// ALERT DLQMessagesIncreasing
//   IF rate(dlq_messages_total[5m]) > 0
//   FOR 1m
//   LABELS { severity = "critical" }
//   ANNOTATIONS { summary = "Messages being sent to DLQ" }

技术选型对比

常见问题

总结

事件驱动架构是构建可扩展、弹性和松耦合分布式系统的强大范式。Apache Kafka 和 RabbitMQ 提供了不同场景下的消息传递基础设施。事件溯源和 CQRS 为复杂业务逻辑提供了审计和性能优化。Saga 模式解决了分布式事务的挑战。Schema 演进确保系统长期可维护。流处理引擎支持实时分析。全面的测试和监控策略则保障了生产环境的可靠性。根据你的具体需求选择合适的模式和工具组合，从简单开始，随着复杂性增长逐步引入更多模式。