Koa.js 事件驱动架构与 Event Sourcing 实践

发布时间：2026-06-21 | 分类：架构设计

引言

事件驱动架构（EDA）和事件溯源（Event Sourcing）是现代分布式系统的重要设计模式。本文将介绍如何在 Koa.js 中实现事件驱动架构，构建可追溯、可扩展的应用系统。

CQRS 架构

命令查询职责分离模式：

核心代码实现

事件总线实现：

// 事件总线
class EventBus {
  constructor(config = {}) {
    this.subscribers = new Map();
    this.middlewares = [];
    this.eventStore = config.eventStore || new InMemoryEventStore();
    this.logger = config.logger || console;
  }

  // 注册中间件
  use(middleware) {
    this.middlewares.push(middleware);
  }

  // 订阅事件
  subscribe(eventType, handler) {
    if (!this.subscribers.has(eventType)) {
      this.subscribers.set(eventType, []);
    }
    this.subscribers.get(eventType).push(handler);
  }

  // 取消订阅
  unsubscribe(eventType, handler) {
    const handlers = this.subscribers.get(eventType);
    if (handlers) {
      const index = handlers.indexOf(handler);
      if (index > -1) {
        handlers.splice(index, 1);
      }
    }
  }

  // 发布事件
  async publish(event, metadata = {}) {
    const enrichedEvent = {
      ...event,
      eventId: this.generateId(),
      timestamp: new Date().toISOString(),
      metadata
    };

    this.logger.info(`Event published: ${enrichedEvent.eventType}`);

    // 存储事件
    await this.eventStore.append(enrichedEvent);

    // 执行中间件
    let context = { event: enrichedEvent };
    for (const middleware of this.middlewares) {
      context = await middleware(context);
    }

    // 通知订阅者
    const handlers = this.subscribers.get(enrichedEvent.eventType) || [];
    const asyncHandlers = this.subscribers.get('*') || []; // 通配符处理器

    const allHandlers = [...handlers, ...asyncHandlers];

    const results = await Promise.allSettled(
      allHandlers.map(handler =>
        handler(enrichedEvent).catch(err => ({ error: err.message }))
      )
    );

    // 检查失败
    const failures = results.filter(r => r.status === 'rejected');
    if (failures.length > 0) {
      this.logger.error(`${failures.length} handlers failed`);
    }

    return enrichedEvent;
  }

  // 生成唯一ID
  generateId() {
    return `${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
  }
}

// 事件存储器
class EventStore {
  async append(event) {
    throw new Error('must be implemented');
  }

  async getEvents(aggregateId, fromVersion = 0) {
    throw new Error('must be implemented');
  }

  async getAllEvents(fromOffset = 0, limit = 100) {
    throw new Error('must be implemented');
  }
}

// 内存事件存储（开发环境）
class InMemoryEventStore extends EventStore {
  constructor() {
    super();
    this.events = [];
    this.aggregateIndexes = new Map();
  }

  async append(event) {
    this.events.push(event);

    // 建立聚合索引
    if (event.aggregateId) {
      if (!this.aggregateIndexes.has(event.aggregateId)) {
        this.aggregateIndexes.set(event.aggregateId, []);
      }
      this.aggregateIndexes.get(event.aggregateId).push(event);
    }
  }

  async getEvents(aggregateId, fromVersion = 0) {
    const events = this.aggregateIndexes.get(aggregateId) || [];
    return events.filter(e => e.aggregateVersion > fromVersion);
  }

  async getAllEvents(fromOffset = 0, limit = 100) {
    return this.events.slice(fromOffset, fromOffset + limit);
  }
}

//MongoDB 事件存储
class MongoEventStore extends EventStore {
  constructor(mongoClient, database = 'events') {
    super();
    this.collection = mongoClient.db(database).collection('events');
    this.collection.createIndex({ aggregateId: 1, aggregateVersion: 1 });
    this.collection.createIndex({ timestamp: 1 });
  }

  async append(event) {
    await this.collection.insertOne({
      ...event,
      _savedAt: new Date()
    });
  }

  async getEvents(aggregateId, fromVersion = 0) {
    return await this.collection.find({
      aggregateId,
      aggregateVersion: { $gt: fromVersion }
    }).sort({ aggregateVersion: 1 }).toArray();
  }

  async getAllEvents(fromOffset = 0, limit = 100) {
    return await this.collection.find({})
      .sort({ timestamp: 1 })
      .skip(fromOffset)
      .limit(limit)
      .toArray();
  }
}

// 命令处理器
class CommandHandler {
  constructor(eventBus, aggregateFactory) {
    this.eventBus = eventBus;
    this.aggregateFactory = aggregateFactory;
    this.handlers = new Map();
  }

  // 注册命令处理器
  register(commandType, handler) {
    this.handlers.set(commandType, handler);
  }

  // 处理命令
  async handle(command) {
    const handler = this.handlers.get(command.type);
    if (!handler) {
      throw new Error(`No handler for command: ${command.type}`);
    }

    // 获取聚合根
    let aggregate;
    if (command.aggregateId) {
      aggregate = await this.aggregateFactory.load(
        command.aggregateId,
        command.aggregateType
      );
    }

    // 执行命令处理
    const result = await handler(aggregate, command);

    // 发布事件
    if (result.events && result.events.length > 0) {
      for (const event of result.events) {
        await this.eventBus.publish(event, {
          commandId: command.commandId,
          aggregateId: command.aggregateId,
          userId: command.userId
        });
      }
    }

    // 保存聚合
    if (aggregate && result.events.length > 0) {
      await this.aggregateFactory.save(aggregate);
    }

    return result;
  }
}

// 聚合根
class AggregateRoot {
  constructor(id, type) {
    this.id = id;
    this.type = type;
    this.version = 0;
    this.uncommittedEvents = [];
    this.state = this.getInitialState();
  }

  getInitialState() {
    return {};
  }

  // 应用事件更新状态
  apply(event) {
    this.version++;
    const handler = `apply${this.capitalize(event.eventType)}`;
    if (this[handler]) {
      this[handler](event);
    }
  }

  // 收集未提交事件
  collectEvents() {
    const events = [...this.uncommittedEvents];
    this.uncommittedEvents = [];
    return events;
  }

  // 添加未提交事件
  addUncommittedEvent(event) {
    this.uncommittedEvents.push({
      ...event,
      aggregateId: this.id,
      aggregateType: this.type,
      aggregateVersion: this.version + 1
    });
  }

  capitalize(str) {
    return str.charAt(0).toUpperCase() + str.slice(1);
  }
}

Event Sourcing 实现

事件溯源与状态重建：

// 聚合工厂
class AggregateFactory {
  constructor(eventStore) {
    this.eventStore = eventStore;
    this.aggregates = new Map();
  }

  // 注册聚合类型
  register(type, aggregateClass) {
    this.aggregates.set(type, aggregateClass);
  }

  // 加载聚合
  async load(id, type) {
    const AggregateClass = this.aggregates.get(type);
    if (!AggregateClass) {
      throw new Error(`Unknown aggregate type: ${type}`);
    }

    const aggregate = new AggregateClass(id);
    const events = await this.eventStore.getEvents(id);

    // 重放事件重建状态
    for (const event of events) {
      aggregate.apply(event);
    }

    return aggregate;
  }

  // 保存聚合
  async save(aggregate) {
    const events = aggregate.collectEvents();
    for (const event of events) {
      await this.eventStore.append(event);
    }
  }
}

// 订单聚合示例
class OrderAggregate extends AggregateRoot {
  constructor(orderId) {
    super(orderId, 'Order');
  }

  getInitialState() {
    return {
      customerId: null,
      items: [],
      totalAmount: 0,
      status: 'pending',
      createdAt: null,
      updatedAt: null
    };
  }

  // 创建订单
  createOrder(command) {
    this.addUncommittedEvent({
      eventType: 'OrderCreated',
      data: {
        customerId: command.customerId,
        items: command.items,
        totalAmount: command.totalAmount
      }
    });
  }

  // 应用订单创建事件
  applyOrderCreated(event) {
    this.state.customerId = event.data.customerId;
    this.state.items = event.data.items;
    this.state.totalAmount = event.data.totalAmount;
    this.state.status = 'created';
    this.state.createdAt = event.timestamp;
    this.state.updatedAt = event.timestamp;
  }

  // 添加订单项
  addItem(command) {
    if (this.state.status !== 'created' && this.state.status !== 'pending') {
      throw new Error('Cannot add items to order in current status');
    }

    this.addUncommittedEvent({
      eventType: 'ItemAdded',
      data: {
        itemId: command.itemId,
        productId: command.productId,
        quantity: command.quantity,
        price: command.price
      }
    });
  }

  applyItemAdded(event) {
    this.state.items.push({
      itemId: event.data.itemId,
      productId: event.data.productId,
      quantity: event.data.quantity,
      price: event.data.price
    });
    this.state.totalAmount += event.data.price * event.data.quantity;
    this.state.updatedAt = event.timestamp;
  }

  // 确认订单
  confirmOrder() {
    if (this.state.status !== 'created') {
      throw new Error('Order cannot be confirmed in current status');
    }

    this.addUncommittedEvent({
      eventType: 'OrderConfirmed',
      data: {}
    });
  }

  applyOrderConfirmed(event) {
    this.state.status = 'confirmed';
    this.state.updatedAt = event.timestamp;
  }

  // 完成订单
  completeOrder() {
    if (this.state.status !== 'confirmed') {
      throw new Error('Order cannot be completed in current status');
    }

    this.addUncommittedEvent({
      eventType: 'OrderCompleted',
      data: {}
    });
  }

  applyOrderCompleted(event) {
    this.state.status = 'completed';
    this.state.updatedAt = event.timestamp;
  }
}

// 查询模型管理器
class ReadModelManager {
  constructor(eventBus) {
    this.projectors = new Map();
    this.queryModels = {};
    eventBus.subscribe('*', this.handleEvent.bind(this));
  }

  // 注册投影器
  register(projectionName, projector) {
    this.projectors.set(projectionName, projector);
    this.queryModels[projectionName] = projector.getInitialState();
  }

  // 处理事件
  async handleEvent(event) {
    for (const [name, projector] of this.projectors) {
      try {
        await projector.project(event, this.queryModels[name]);
      } catch (error) {
        console.error(`Projection ${name} failed:`, error);
      }
    }
  }

  // 获取查询模型
  getQueryModel(name) {
    return this.queryModels[name];
  }
}

// 订单查询模型投影器
class OrderQueryProjector {
  getInitialState() {
    return {
      orders: new Map(),
      ordersByCustomer: new Map(),
      ordersByStatus: new Map()
    };
  }

  async project(event, state) {
    switch (event.eventType) {
      case 'OrderCreated':
        state.orders.set(event.aggregateId, {
          id: event.aggregateId,
          customerId: event.data.customerId,
          items: event.data.items,
          totalAmount: event.data.totalAmount,
          status: 'created',
          createdAt: event.timestamp
        });

        // 按客户索引
        if (!state.ordersByCustomer.has(event.data.customerId)) {
          state.ordersByCustomer.set(event.data.customerId, []);
        }
        state.ordersByCustomer.get(event.data.customerId).push(event.aggregateId);

        // 按状态索引
        if (!state.ordersByStatus.has('created')) {
          state.ordersByStatus.set('created', []);
        }
        state.ordersByStatus.get('created').push(event.aggregateId);
        break;

      case 'OrderConfirmed':
        const order1 = state.orders.get(event.aggregateId);
        if (order1) {
          order1.status = 'confirmed';
          this.moveStatusIndex(state.ordersByStatus, event.aggregateId, 'created', 'confirmed');
        }
        break;

      case 'OrderCompleted':
        const order2 = state.orders.get(event.aggregateId);
        if (order2) {
          order2.status = 'completed';
          this.moveStatusIndex(state.ordersByStatus, event.aggregateId, 'confirmed', 'completed');
        }
        break;
    }
  }

  moveStatusIndex(indexMap, orderId, fromStatus, toStatus) {
    if (indexMap.has(fromStatus)) {
      const list = indexMap.get(fromStatus);
      const idx = list.indexOf(orderId);
      if (idx > -1) list.splice(idx, 1);
    }
    if (!indexMap.has(toStatus)) {
      indexMap.set(toStatus, []);
    }
    indexMap.get(toStatus).push(orderId);
  }
}

最佳实践建议

• 事件设计：使用描述性的事件名，记录足够的状态变化信息
• 幂等处理：事件处理器需支持幂等，避免重复处理导致数据错误
• 版本管理：为事件添加版本号，便于架构演进
• 快照策略：大聚合定期创建快照，提升加载性能
• 最终一致：读写模型分离，需接受最终一致性

总结

事件驱动架构与 Event Sourcing 为应用带来：

• 完整追溯：记录所有状态变化，支持任意时间点回溯
• CQRS 分离：读写分离，优化各自性能
• 可扩展：新增事件类型不影响现有逻辑
• 审计友好：完整的事件日志满足审计需求

通过在 Koa.js 中实现事件驱动架构，可以构建更健壮、可维护的应用系统。