事件驱动架构：构建松耦合、高扩展系统的核心范式

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// 传统请求/响应模式 - 紧耦合
class OrderController {
  async createOrder(dto: OrderDTO): Promise<Order> {
    const order = await this.orderService.create(dto);

    // 直接调用其他服务 - 强依赖
    await this.inventoryService.reserve(order);
    await this.paymentService.charge(order);
    await this.notificationService.send(order);

    return order;
  }
}

// 事件驱动模式 - 松耦合
class OrderController {
  async createOrder(dto: OrderDTO): Promise<Order> {
    const order = await this.orderService.create(dto);

    // 发布事件 - 其他服务订阅响应
    await this.eventBus.publish('order.created', {
      orderId: order.id,
      userId: order.userId,
      items: order.items,
      total: order.total
    });

    return order; // 立即返回，不等待其他服务
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// ✅ 好的事件命名 - 过去式、描述事实
interface Events {
  'user.created': UserCreatedEvent;
  'order.paid': OrderPaidEvent;
  'inventory.reserved': InventoryReservedEvent;
  'payment.failed': PaymentFailedEvent;
}

// ❌ 避免的事件命名 - 命令式、描述意图
interface BadEvents {
  'createUser': CreateUserEvent;  // 这是命令，不是事件
  'orderPay': OrderPayEvent;      // 不清晰
  'reserveInventory': ReserveEvent; // 动作而非事实
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

// 订单创建事件
interface OrderCreatedEvent {
  // 事件元数据
  eventId: string;        // 唯一事件ID
  eventType: string;       // 'order.created'
  timestamp: number;       // 事件时间戳
  correlationId: string;   // 关联ID（追踪用）

  // 业务数据
  aggregateId: string;     // 聚合根ID（订单ID）
  aggregateVersion: number; // 聚合版本号（乐观锁）

  // 事件数据
  data: {
    userId: string;
    items: Array<{
      productId: string;
      quantity: number;
      price: number;
    }>;
    totalAmount: number;
    shippingAddress: Address;
  };
}

// 事件发布
async function publishOrderCreated(order: Order): Promise<void> {
  const event: OrderCreatedEvent = {
    eventId: uuidv4(),
    eventType: 'order.created',
    timestamp: Date.now(),
    correlationId: order.correlationId,
    aggregateId: order.id,
    aggregateVersion: order.version,
    data: {
      userId: order.userId,
      items: order.items,
      totalAmount: order.total,
      shippingAddress: order.shippingAddress
    }
  };

  await eventBus.publish('order.created', event);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// 事件总线实现
import { EventEmitter } from 'events';

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

  // 订阅事件
  subscribe(
    eventType: string,
    handler: Handler
  ): Unsubscribe {
    const handlers = this.subscribers.get(eventType) || [];
    handlers.push(handler);
    this.subscribers.set(eventType, handlers);

    this.emitter.on(eventType, handler);

    // 返回取消订阅函数
    return () => {
      this.emitter.off(eventType, handler);
      const index = handlers.indexOf(handler);
      if (index > -1) {
        handlers.splice(index, 1);
      }
    };
  }

  // 发布事件
  async publish(eventType: string, event: Event): Promise<void> {
    const handlers = this.subscribers.get(eventType) || [];

    // 并行处理所有订阅者
    await Promise.all(
      handlers.map(async (handler) => {
        try {
          await handler(event);
        } catch (error) {
          console.error(`Handler error for ${eventType}:`, error);
          // 错误隔离 - 单个处理器失败不影响其他
        }
      })
    );
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106

// 事件存储
interface EventStore {
  append(aggregateId: string, events: Event[]): Promise<void>;
  getEvents(aggregateId: string): Promise<Event[]>;
  getEventsFromVersion(aggregateId string, version: number): Promise<Event[]>;
}

// 实现
class PostgresEventStore implements EventStore {
  constructor(private db: Database) {}

  async append(aggregateId: string, events: Event[]): Promise<void> {
    await this.db.transaction(async (tx) => {
      // 检查版本冲突（乐观锁）
      const current = await this.getCurrentVersion(tx, aggregateId);
      if (current !== events[0].aggregateVersion - 1) {
        throw new ConcurrencyError();
      }

      // 批量插入事件
      for (const event of events) {
        await tx.insert('events', {
          id: event.eventId,
          aggregate_id: aggregateId,
          event_type: event.eventType,
          data: JSON.stringify(event.data),
          version: event.aggregateVersion,
          timestamp: event.timestamp
        });
      }
    });
  }

  async getEvents(aggregateId: string): Promise<Event[]> {
    const rows = await this.db.query(
      `SELECT * FROM events
       WHERE aggregate_id = $1
       ORDER BY version ASC`,
      [aggregateId]
    );

    return rows.map(row => ({
      eventId: row.id,
      eventType: row.event_type,
      data: JSON.parse(row.data),
      aggregateVersion: row.version,
      timestamp: row.timestamp
    }));
  }

  // 从事件重建聚合状态
  async rebuildAggregate<T>(
    aggregateId: string,
    initialState: T,
    applyEvent: (state: T, event: Event) => T
  ): Promise<T> {
    const events = await this.getEvents(aggregateId);
    return events.reduce(
      (state, event) => applyEvent(state, event),
      initialState
    );
  }
}

// 使用示例：订单聚合
interface OrderState {
  id: string;
  status: 'created' | 'paid' | 'shipped' | 'cancelled';
  items: OrderItem[];
  total: number;
}

function applyOrderEvent(state: OrderState, event: Event): OrderState {
  switch (event.eventType) {
    case 'order.created':
      return {
        ...state,
        id: event.aggregateId,
        status: 'created',
        items: event.data.items,
        total: event.data.total
      };

    case 'order.paid':
      return {
        ...state,
        status: 'paid'
      };

    case 'order.shipped':
      return {
        ...state,
        status: 'shipped'
      };

    default:
      return state;
  }
}

// 重建订单状态
const orderState = await eventStore.rebuildAggregate(
  orderId,
  {} as OrderState,
  applyOrderEvent
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

// 命令端：处理写操作
class OrderCommandHandler {
  constructor(
    private eventStore: EventStore,
    private eventBus: EventBus
  ) {}

  async createOrder(command: CreateOrderCommand): Promise<string> {
    const orderId = uuidv4();

    // 生成事件
    const event: OrderCreatedEvent = {
      eventId: uuidv4(),
      eventType: 'order.created',
      aggregateId: orderId,
      aggregateVersion: 1,
      timestamp: Date.now(),
      data: command.data
    };

    // 持久化事件
    await this.eventStore.append(orderId, [event]);

    // 发布事件
    await this.eventBus.publish('order.created', event);

    return orderId;
  }
}

// 查询端：处理读操作
class OrderQueryHandler {
  constructor(private readDb: Database) {}

  async getOrder(orderId: string): Promise<OrderView> {
    return await this.readDb.query(
      'SELECT * FROM order_views WHERE id = $1',
      [orderId]
    );
  }

  async listOrders(filter: OrderFilter): Promise<OrderView[]> {
    return await this.readDb.query(
      'SELECT * FROM order_views WHERE status = $1 ORDER BY created_at DESC',
      [filter.status]
    );
  }
}

// 事件投影：更新读模型
class OrderProjection {
  constructor(
    private eventBus: EventBus,
    private readDb: Database
  ) {
    this.setupSubscriptions();
  }

  private setupSubscriptions(): void {
    this.eventBus.subscribe('order.created', async (event) => {
      await this.readDb.insert('order_views', {
        id: event.aggregateId,
        status: 'created',
        items: event.data.items,
        total: event.data.total,
        created_at: new Date(event.timestamp)
      });
    });

    this.eventBus.subscribe('order.paid', async (event) => {
      await this.readDb.update(
        'order_views',
        { status: 'paid' },
        { id: event.aggregateId }
      );
    });
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// RabbitMQ事件发布
import { connect, Channel } from 'amqplib';

class RabbitMQEventBus {
  private channel: Channel;

  async connect(): Promise<void> {
    const connection = await connect('amqp://localhost');
    this.channel = await connection.createChannel();
  }

  // 发布事件到交换机
  async publish(
    exchange: string,
    routingKey: string,
    event: Event
  ): Promise<void> {
    await this.channel.assertExchange(exchange, 'topic', {
      durable: true
    });

    const message = Buffer.from(JSON.stringify(event));

    this.channel.publish(
      exchange,
      routingKey,
      message,
      {
        persistent: true,  // 持久化
        messageId: event.eventId,
        timestamp: event.timestamp
      }
    );
  }

  // 订阅事件
  async subscribe(
    exchange: string,
    routingKey: string,
    queue: string,
    handler: (event: Event) => void
  ): Promise<void> {
    await this.channel.assertExchange(exchange, 'topic', {
      durable: true
    });

    await this.channel.assertQueue(queue, { durable: true });
    await this.channel.bindQueue(queue, exchange, routingKey);

    this.channel.consume(queue, async (msg) => {
      if (msg) {
        try {
          const event = JSON.parse(msg.content.toString());
          await handler(event);
          this.channel.ack(msg);
        } catch (error) {
          console.error('Event processing error:', error);
          this.channel.nack(msg, false, true); // 重新入队
        }
      }
    });
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

// Kafka事件流
import { Kafka } from 'kafkajs';

class KafkaEventBus {
  private kafka: Kafka;
  private producer: any;

  constructor() {
    this.kafka = new Kafka({
      clientId: 'order-service',
      brokers: ['localhost:9092']
    });
    this.producer = this.kafka.producer();
  }

  async publish(topic: string, event: Event): Promise<void> {
    await this.producer.send({
      topic,
      messages: [
        {
          key: event.aggregateId, // 按聚合分区
          value: JSON.stringify(event),
          headers: {
            eventType: event.eventType,
            eventId: event.eventId
          }
        }
      ]
    });
  }

  async subscribe(
    topic: string,
    groupId: string,
    handler: (event: Event) => void
  ): Promise<void> {
    const consumer = this.kafka.consumer({ groupId });

    await consumer.connect();
    await consumer.subscribe({ topic, fromBeginning: false });

    await consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        try {
          const event = JSON.parse(message.value.toString());
          await handler(event);
        } catch (error) {
          console.error('Message processing error:', error);
        }
      }
    });
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

// 多个消费者实例竞争处理消息
class OrderProcessor {
  private isProcessing = false;

  async start(): Promise<void> {
    while (true) {
      try {
        // 从队列获取消息
        const message = await this.messageQueue.receive('orders');

        if (!message) {
          await this.sleep(1000);
          continue;
        }

        // 处理消息
        await this.processOrder(message);

        // 确认消息
        await this.messageQueue.ack(message);
      } catch (error) {
        console.error('Processing error:', error);
      }
    }
  }

  private async processOrder(message: Message): Promise<void> {
    const event: OrderCreatedEvent = JSON.parse(message.body);

    // 业务逻辑
    await this.inventoryService.reserve(event.data.items);
    await this.notificationService.send(event);
  }

  private sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// 确保重复消息不会产生副作用
class IdempotentHandler {
  private processedEvents = new RedisCache();

  async handle(event: Event): Promise<void> {
    // 检查是否已处理
    const key = `processed:${event.eventId}`;
    const exists = await this.processedEvents.get(key);

    if (exists) {
      console.log(`Event ${event.eventId} already processed, skipping`);
      return;
    }

    // 处理事件
    await this.processEvent(event);

    // 标记已处理（设置过期时间）
    await this.processedEvents.set(key, '1', { EX: 86400 });
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

// 事件演进策略
interface EventV1 {
  eventType: 'user.created';
  data: {
    name: string;
    email: string;
  };
}

interface EventV2 {
  eventType: 'user.created';
  version: 2;
  data: {
    firstName: string;
    lastName: string;
    email: string;
    phone?: string;
  };
}

// 事件升级器
class EventUpgrader {
  private upgraders = new Map<string, (event: any) => any>();

  constructor() {
    this.upgraders.set('user.created', this.upgradeUserCreated);
  }

  upgrade(event: any): any {
    const upgrader = this.upgraders.get(event.eventType);
    if (upgrader && (!event.version || event.version < 2)) {
      return upgrader(event);
    }
    return event;
  }

  private upgradeUserCreated(v1: any): any {
    // V1 -> V2
    const names = v1.data.name.split(' ');
    return {
      ...v1,
      version: 2,
      data: {
        firstName: names[0],
        lastName: names.slice(1).join(' '),
        email: v1.data.email
      }
    };
  }
}