event-driven-architect

事件驱动架构师

使用事件驱动模式构建解耦、可扩展的系统。

核心工作流

1. 识别领域事件：定义发生了什么
2. 设计事件架构：构建事件负载
3. 实现事件总线：发布和订阅
4. 添加事件处理器：响应事件
5. 考虑 CQRS：分离读写操作
6. 启用事件溯源：存储事件历史记录

事件基础

事件结构

// events/base.ts
export interface DomainEvent<T = unknown> {
  id: string;
  type: string;
  aggregateId: string;
  aggregateType: string;
  payload: T;
  metadata: {
    timestamp: Date;
    version: number;
    correlationId?: string;
    causationId?: string;
    userId?: string;
  };
}

// Type-safe event creator
export function createEvent<T>(
  type: string,
  aggregateType: string,
  aggregateId: string,
  payload: T,
  metadata?: Partial<DomainEvent['metadata']>
): DomainEvent<T> {
  return {
    id: crypto.randomUUID(),
    type,
    aggregateType,
    aggregateId,
    payload,
    metadata: {
      timestamp: new Date(),
      version: 1,
      ...metadata,
    },
  };
}

定义领域事件

// events/order.events.ts
export interface OrderCreatedPayload {
  customerId: string;
  items: Array<{
    productId: string;
    quantity: number;
    price: number;
  }>;
  totalAmount: number;
  shippingAddress: Address;
}

export interface OrderPaidPayload {
  paymentId: string;
  amount: number;
  method: 'card' | 'bank' | 'wallet';
}

export interface OrderShippedPayload {
  trackingNumber: string;
  carrier: string;
  estimatedDelivery: string;
}

export interface OrderCancelledPayload {
  reason: string;
  cancelledBy: string;
  refundAmount?: number;
}

// Event types
export type OrderEvent =
  | DomainEvent<OrderCreatedPayload> & { type: 'OrderCreated' }
  | DomainEvent<OrderPaidPayload> & { type: 'OrderPaid' }
  | DomainEvent<OrderShippedPayload> & { type: 'OrderShipped' }
  | DomainEvent<OrderCancelledPayload> & { type: 'OrderCancelled' };

// Event creators
export const OrderEvents = {
  created: (orderId: string, payload: OrderCreatedPayload) =>
    createEvent('OrderCreated', 'Order', orderId, payload),

  paid: (orderId: string, payload: OrderPaidPayload) =>
    createEvent('OrderPaid', 'Order', orderId, payload),

  shipped: (orderId: string, payload: OrderShippedPayload) =>
    createEvent('OrderShipped', 'Order', orderId, payload),

  cancelled: (orderId: string, payload: OrderCancelledPayload) =>
    createEvent('OrderCancelled', 'Order', orderId, payload),
};

事件总线

内存事件总线

// events/event-bus.ts
import { EventEmitter } from 'events';
import { DomainEvent } from './base';

type EventHandler<T = unknown> = (event: DomainEvent<T>) => Promise<void>;

class EventBus {
  private emitter = new EventEmitter();
  private handlers = new Map<string, EventHandler[]>();

  async publish<T>(event: DomainEvent<T>): Promise<void> {
    console.log(`Publishing event: ${event.type}`, event);

    // Store event (for event sourcing)
    await this.storeEvent(event);

    // Emit to handlers
    this.emitter.emit(event.type, event);
    this.emitter.emit('*', event); // Wildcard for all events
  }

  async publishAll(events: DomainEvent[]): Promise<void> {
    for (const event of events) {
      await this.publish(event);
    }
  }

  subscribe<T>(eventType: string, handler: EventHandler<T>): () => void {
    const wrappedHandler = async (event: DomainEvent<T>) => {
      try {
        await handler(event);
      } catch (error) {
        console.error(`Error handling ${eventType}:`, error);
        // Could emit to dead letter queue here
      }
    };

    this.emitter.on(eventType, wrappedHandler);

    // Return unsubscribe function
    return () => {
      this.emitter.off(eventType, wrappedHandler);
    };
  }

  subscribeAll(handler: EventHandler): () => void {
    return this.subscribe('*', handler);
  }

  private async storeEvent(event: DomainEvent): Promise<void> {
    await db.event.create({
      data: {
        id: event.id,
        type: event.type,
        aggregateId: event.aggregateId,
        aggregateType: event.aggregateType,
        payload: event.payload as any,
        metadata: event.metadata as any,
        createdAt: event.metadata.timestamp,
      },
    });
  }
}

export const eventBus = new EventBus();

基于 Redis 的事件总线

// events/redis-event-bus.ts
import { Redis } from 'ioredis';
import { DomainEvent } from './base';

const publisher = new Redis(process.env.REDIS_URL!);
const subscriber = new Redis(process.env.REDIS_URL!);

class RedisEventBus {
  private handlers = new Map<string, Set<(event: DomainEvent) => Promise<void>>>();

  constructor() {
    subscriber.on('message', async (channel, message) => {
      const event = JSON.parse(message) as DomainEvent;
      const handlers = this.handlers.get(channel) || new Set();

      for (const handler of handlers) {
        try {
          await handler(event);
        } catch (error) {
          console.error(`Error handling ${event.type}:`, error);
        }
      }
    });
  }

  async publish(event: DomainEvent): Promise<void> {
    const channel = `events:${event.type}`;
    await publisher.publish(channel, JSON.stringify(event));

    // Also store in stream for replay
    await publisher.xadd(
      `stream:${event.aggregateType}`,
      '*',
      'event',
      JSON.stringify(event)
    );
  }

  subscribe(eventType: string, handler: (event: DomainEvent) => Promise<void>): () => void {
    const channel = `events:${eventType}`;

    if (!this.handlers.has(channel)) {
      this.handlers.set(channel, new Set());
      subscriber.subscribe(channel);
    }

    this.handlers.get(channel)!.add(handler);

    return () => {
      this.handlers.get(channel)?.delete(handler);
    };
  }
}

export const eventBus = new RedisEventBus();

事件处理器

处理器注册

// handlers/order.handlers.ts
import { eventBus } from '../events/event-bus';
import { OrderEvent } from '../events/order.events';

// Email notification on order created
eventBus.subscribe<OrderCreatedPayload>('OrderCreated', async (event) => {
  await emailService.send({
    to: await getUserEmail(event.payload.customerId),
    template: 'order-confirmation',
    data: {
      orderId: event.aggregateId,
      items: event.payload.items,
      total: event.payload.totalAmount,
    },
  });
});

// Update inventory on order created
eventBus.subscribe<OrderCreatedPayload>('OrderCreated', async (event) => {
  for (const item of event.payload.items) {
    await inventoryService.reserve(item.productId, item.quantity);
  }
});

// Analytics tracking
eventBus.subscribe<OrderPaidPayload>('OrderPaid', async (event) => {
  await analytics.track('order_completed', {
    orderId: event.aggregateId,
    amount: event.payload.amount,
    paymentMethod: event.payload.method,
  });
});

// Notify shipping on order paid
eventBus.subscribe<OrderPaidPayload>('OrderPaid', async (event) => {
  await shippingService.createShipment(event.aggregateId);
});

// Handle cancellation
eventBus.subscribe<OrderCancelledPayload>('OrderCancelled', async (event) => {
  // Release inventory
  const order = await orderRepository.findById(event.aggregateId);
  for (const item of order.items) {
    await inventoryService.release(item.productId, item.quantity);
  }

  // Process refund
  if (event.payload.refundAmount) {
    await paymentService.refund(event.aggregateId, event.payload.refundAmount);
  }

  // Send cancellation email
  await emailService.send({
    to: await getUserEmail(order.customerId),
    template: 'order-cancelled',
    data: {
      orderId: event.aggregateId,
      reason: event.payload.reason,
    },
  });
});

事件溯源

包含事件的聚合

// aggregates/order.aggregate.ts
import { DomainEvent } from '../events/base';
import { OrderEvents, OrderCreatedPayload, OrderPaidPayload } from '../events/order.events';

interface OrderItem {
  productId: string;
  quantity: number;
  price: number;
}

type OrderStatus = 'pending' | 'paid' | 'shipped' | 'delivered' | 'cancelled';

export class OrderAggregate {
  private _id: string;
  private _status: OrderStatus = 'pending';
  private _items: OrderItem[] = [];
  private _totalAmount: number = 0;
  private _customerId: string = '';
  private _version: number = 0;

  private uncommittedEvents: DomainEvent[] = [];

  get id() { return this._id; }
  get status() { return this._status; }
  get items() { return [...this._items]; }
  get version() { return this._version; }

  constructor(id?: string) {
    this._id = id || crypto.randomUUID();
  }

  // Command: Create order
  static create(customerId: string, items: OrderItem[], shippingAddress: Address): OrderAggregate {
    const order = new OrderAggregate();
    const totalAmount = items.reduce((sum, item) => sum + item.price * item.quantity, 0);

    order.apply(
      OrderEvents.created(order._id, {
        customerId,
        items,
        totalAmount,
        shippingAddress,
      })
    );

    return order;
  }

  // Command: Pay order
  pay(paymentId: string, amount: number, method: 'card' | 'bank' | 'wallet'): void {
    if (this._status !== 'pending') {
      throw new Error('Order cannot be paid in current status');
    }

    if (amount !== this._totalAmount) {
      throw new Error('Payment amount does not match order total');
    }

    this.apply(
      OrderEvents.paid(this._id, { paymentId, amount, method })
    );
  }

  // Command: Cancel order
  cancel(reason: string, cancelledBy: string): void {
    if (['shipped', 'delivered', 'cancelled'].includes(this._status)) {
      throw new Error('Order cannot be cancelled in current status');
    }

    const refundAmount = this._status === 'paid' ? this._totalAmount : undefined;

    this.apply(
      OrderEvents.cancelled(this._id, { reason, cancelledBy, refundAmount })
    );
  }

  // Apply event and track for persistence
  private apply(event: DomainEvent): void {
    this.applyEvent(event);
    this.uncommittedEvents.push(event);
  }

  // Apply event to state (used for replay too)
  private applyEvent(event: DomainEvent): void {
    switch (event.type) {
      case 'OrderCreated':
        const created = event.payload as OrderCreatedPayload;
        this._customerId = created.customerId;
        this._items = created.items;
        this._totalAmount = created.totalAmount;
        this._status = 'pending';
        break;

      case 'OrderPaid':
        this._status = 'paid';
        break;

      case 'OrderShipped':
        this._status = 'shipped';
        break;

      case 'OrderCancelled':
        this._status = 'cancelled';
        break;
    }

    this._version++;
  }

  // Get and clear uncommitted events
  getUncommittedEvents(): DomainEvent[] {
    const events = [...this.uncommittedEvents];
    this.uncommittedEvents = [];
    return events;
  }

  // Rebuild from events
  static fromEvents(events: DomainEvent[]): OrderAggregate {
    if (events.length === 0) {
      throw new Error('Cannot rebuild aggregate from empty events');
    }

    const order = new OrderAggregate(events[0].aggregateId);

    for (const event of events) {
      order.applyEvent(event);
    }

    return order;
  }
}

事件存储库

// repositories/event-store.repository.ts
import { db } from '../lib/db';
import { DomainEvent } from '../events/base';
import { eventBus } from '../events/event-bus';

export class EventStoreRepository<T extends { id: string; getUncommittedEvents(): DomainEvent[] }> {
  constructor(
    private aggregateType: string,
    private reconstruct: (events: DomainEvent[]) => T
  ) {}

  async save(aggregate: T): Promise<void> {
    const events = aggregate.getUncommittedEvents();

    if (events.length === 0) return;

    // Store events
    await db.event.createMany({
      data: events.map((event) => ({
        id: event.id,
        type: event.type,
        aggregateId: event.aggregateId,
        aggregateType: event.aggregateType,
        payload: event.payload as any,
        metadata: event.metadata as any,
        createdAt: event.metadata.timestamp,
      })),
    });

    // Publish events
    await eventBus.publishAll(events);
  }

  async findById(id: string): Promise<T | null> {
    const events = await db.event.findMany({
      where: {
        aggregateId: id,
        aggregateType: this.aggregateType,
      },
      orderBy: { createdAt: 'asc' },
    });

    if (events.length === 0) return null;

    return this.reconstruct(
      events.map((e) => ({
        id: e.id,
        type: e.type,
        aggregateId: e.aggregateId,
        aggregateType: e.aggregateType,
        payload: e.payload,
        metadata: e.metadata as any,
      }))
    );
  }

  async getEvents(aggregateId: string, fromVersion?: number): Promise<DomainEvent[]> {
    const events = await db.event.findMany({
      where: {
        aggregateId,
        aggregateType: this.aggregateType,
        ...(fromVersion && {
          metadata: { path: ['version'], gte: fromVersion },
        }),
      },
      orderBy: { createdAt: 'asc' },
    });

    return events.map((e) => ({
      id: e.id,
      type: e.type,
      aggregateId: e.aggregateId,
      aggregateType: e.aggregateType,
      payload: e.payload,
      metadata: e.metadata as any,
    }));
  }
}

// Usage
export const orderRepository = new EventStoreRepository(
  'Order',
  OrderAggregate.fromEvents
);

CQRS 模式

分离命令和查询

// commands/create-order.command.ts
export interface CreateOrderCommand {
  customerId: string;
  items: Array<{ productId: string; quantity: number }>;
  shippingAddress: Address;
}

// Command handler
export async function handleCreateOrder(command: CreateOrderCommand): Promise<string> {
  // Validate
  const customer = await customerRepository.findById(command.customerId);
  if (!customer) throw new Error('Customer not found');

  // Get product prices
  const items = await Promise.all(
    command.items.map(async (item) => {
      const product = await productRepository.findById(item.productId);
      return {
        productId: item.productId,
        quantity: item.quantity,
        price: product.price,
      };
    })
  );

  // Create aggregate and save
  const order = OrderAggregate.create(
    command.customerId,
    items,
    command.shippingAddress
  );

  await orderRepository.save(order);

  return order.id;
}

// queries/order.queries.ts
// Read model - denormalized for fast queries
export interface OrderReadModel {
  id: string;
  status: string;
  customerName: string;
  customerEmail: string;
  items: Array<{
    productName: string;
    quantity: number;
    price: number;
  }>;
  totalAmount: number;
  createdAt: Date;
  paidAt?: Date;
  shippedAt?: Date;
}

// Query handler
export async function getOrderById(orderId: string): Promise<OrderReadModel | null> {
  return db.orderReadModel.findUnique({
    where: { id: orderId },
  });
}

export async function getOrdersByCustomer(customerId: string): Promise<OrderReadModel[]> {
  return db.orderReadModel.findMany({
    where: { customerId },
    orderBy: { createdAt: 'desc' },
  });
}

读模型投影器

// projectors/order.projector.ts
import { eventBus } from '../events/event-bus';

// Project events to read model
eventBus.subscribe('OrderCreated', async (event) => {
  const { customerId, items, totalAmount } = event.payload;
  const customer = await db.customer.findUnique({ where: { id: customerId } });

  await db.orderReadModel.create({
    data: {
      id: event.aggregateId,
      status: 'pending',
      customerId,
      customerName: customer.name,
      customerEmail: customer.email,
      items: await enrichItems(items),
      totalAmount,
      createdAt: event.metadata.timestamp,
    },
  });
});

eventBus.subscribe('OrderPaid', async (event) => {
  await db.orderReadModel.update({
    where: { id: event.aggregateId },
    data: {
      status: 'paid',
      paidAt: event.metadata.timestamp,
    },
  });
});

eventBus.subscribe('OrderShipped', async (event) => {
  await db.orderReadModel.update({
    where: { id: event.aggregateId },
    data: {
      status: 'shipped',
      shippedAt: event.metadata.timestamp,
      trackingNumber: event.payload.trackingNumber,
    },
  });
});

最佳实践

1. 不可变事件：绝不修改已存储的事件
2. 描述性事件名称：使用过去时（OrderCreated，而不是 CreateOrder）
3. 包含所有上下文：事件应自成一体
4. 事件版本控制：处理架构演进
5. 幂等处理器：妥善处理重复事件
6. 分离关注点：命令用于修改，查询用于读取
7. 事件版本管理：支持向后兼容
8. 死信队列：处理失败事件

输出检查清单

每个事件驱动系统都应包含：

• 定义明确的领域事件
• 类型安全的事件负载
• 事件总线（内存或分布式）
• 带错误处理的事件处理器
• 用于持久化的事件存储
• 使用事件溯源的聚合（如有需要）
• CQRS 分离（如有需要）
• 读模型投影器
• 死信处理
• 事件重放能力