一、前置思考

多设备协同场景中，事件传递是最基础的需求——手机端点击"分享"按钮，平板端要感知到；PC端修改了文档标题，智慧屏端要更新标题展示。简单地用KVStore轮询监听效率极低，分布式事件总线（Distributed Event Bus）提供了高效的发布-订阅模式。

本文聚焦：

• 分布式事件总线的架构设计
• 事件全局唯一性与顺序保证
• 延迟解耦（Deferred Event）与重放机制
• 事件的TTL生命周期管理

二、核心原理

2.1 事件总线架构

发布者 (Publisher)              订阅者 (Subscriber)
     │                               │
     ├─ publish(e:Event) ──┐        │
     │                      ▼        │
     │                ┌──────────┐   │
     │                │ EventBus │   │
     │                │ ┌──────┐ │   │
     │                │ │Router│──├───┤─→ Topic匹配
     │                │ └──────┘ │   │
     │                │ ┌──────┐ │   │
     │                │ │Queue │ │   │   先入先出
     │                │ └──────┘ │   │
     │                │ ┌──────┐ │   │
     │                │ │Store │ │   │   持久化
     │                │ └──────┘ │   │
     │                └──────────┘   │
     │                               │
     ▼                               ▼
通过软总线跨设备分发            onEvent(Topic)回调

2.2 事件定义

interface DistributedEvent {
  eventId: string;          // 全局唯一ID (UUID v4)
  topic: string;            // 事件主题 (如: "doc:update")
  sourceDeviceId: string;   // 源设备ID
  sourceAppId: string;      // 源应用ID
  timestamp: number;        // 事件发生时间 (UTC毫秒)
  ttl: number;              // 生存时间(ms)，超时丢弃
  priority: number;         // 优先级 0-10
  payload: string;          // 事件载荷(JSON)
  sequenceNumber: number;   // 全局递增序号
}

// 事件生成器
class EventFactory {
  private sequenceCounter: number = 0;
  private deviceId: string;

  constructor(deviceId: string) {
    this.deviceId = deviceId;
  }

  createEvent(
    topic: string,
    payload: string,
    priority: number = 5,
    ttl: number = 30000
  ): DistributedEvent {
    this.sequenceCounter++;
    return {
      eventId: this.generateUUID(),
      topic: topic,
      sourceDeviceId: this.deviceId,
      sourceAppId: 'com.example.app',
      timestamp: Date.now(),
      ttl: ttl,
      priority: priority,
      payload: payload,
      sequenceNumber: this.sequenceCounter
    };
  }

  private generateUUID(): string {
    // 简化UUID生成
    return this.deviceId + '-' + Date.now() + '-' +
      Math.random().toString(36).slice(2, 10);
  }
}

2.3 分布式事件总线核心实现

class DistributedEventBus {
  private kvStore: distributedKVStore.SingleKVStore | null = null;
  private subscribers: Map<string, SubscriberInfo[]> = new Map();
  private eventQueue: DistributedEvent[] = [];
  private readonly MAX_QUEUE_SIZE: number = 1000;
  private readonly EVENT_KEY_PREFIX: string = 'evt:';

  // 发布事件
  async publish(event: DistributedEvent): Promise<void> {
    if (this.kvStore === null) return;

    // 入队（本地队列+KVStore）
    this.enqueue(event);

    // 写入KVStore触发远端同步
    const key: string = this.EVENT_KEY_PREFIX + event.eventId;
    const value: string = JSON.stringify(event);

    await this.kvStore.put(key, value);
    await this.kvStore.sync([], distributedKVStore.SyncMode.PUSH_ONLY);

    // 设置TTL自动清理
    setTimeout(() => {
      this.cleanupEvent(event.eventId);
    }, event.ttl);
  }

  // 订阅
  subscribe(topic: string, callback: (event: DistributedEvent) => void): string {
    const subscriberId: string = topic + '-' + Date.now();
    let subs: SubscriberInfo[] | undefined = this.subscribers.get(topic);
    if (subs === undefined) {
      subs = [];
      this.subscribers.set(topic, subs);
    }
    subs.push({ id: subscriberId, callback: callback, topic: topic });
    return subscriberId;
  }

  // 注销
  unsubscribe(subscriberId: string): void {
    const entries: MapIterator<[string, SubscriberInfo[]]> =
      this.subscribers.entries();
    for (let entry = entries.next(); !entry.done; entry = entries.next()) {
      const topic: string = entry.value[0];
      const subs: SubscriberInfo[] = entry.value[1];
      const newSubs: SubscriberInfo[] = [];
      for (let i: number = 0; i < subs.length; i++) {
        if (subs[i].id !== subscriberId) {
          newSubs.push(subs[i]);
        }
      }
      this.subscribers.set(topic, newSubs);
    }
  }

  // 处理远端事件
  private onRemoteEvent(event: DistributedEvent): void {
    // 检查是否过期
    if (Date.now() - event.timestamp > event.ttl) return;

    // 匹配订阅者
    const subs: SubscriberInfo[] | undefined =
      this.subscribers.get(event.topic);
    if (subs !== undefined) {
      for (let i: number = 0; i < subs.length; i++) {
        subs[i].callback(event);
      }
    }
  }

  private enqueue(event: DistributedEvent): void {
    this.eventQueue.push(event);
    // 按优先队列序排列
    this.eventQueue.sort((a: DistributedEvent, b: DistributedEvent) => {
      if (a.priority !== b.priority) return b.priority - a.priority;
      return a.sequenceNumber - b.sequenceNumber;
    });
    // 队列容量限制
    if (this.eventQueue.length > this.MAX_QUEUE_SIZE) {
      this.eventQueue.shift();
    }
  }

  private async cleanupEvent(eventId: string): Promise<void> {
    if (this.kvStore !== null) {
      await this.kvStore.delete(this.EVENT_KEY_PREFIX + eventId);
    }
  }
}

interface SubscriberInfo {
  id: string;
  topic: string;
  callback: (event: DistributedEvent) => void;
}

三、延迟解耦模式

// 离线设备事件延迟投递
class DeferredEventDelivery {
  private pendingEvents: Map<string, DistributedEvent[]> = new Map();

  // 事件发布时目标设备离线 → 加入pending
  deferEvent(deviceId: string, event: DistributedEvent): void {
    let queue: DistributedEvent[] | undefined =
      this.pendingEvents.get(deviceId);
    if (queue === undefined) {
      queue = [];
      this.pendingEvents.set(deviceId, queue);
    }
    queue.push(event);

    // 限制pending队列大小
    if (queue.length > 100) queue.shift();
  }

  // 设备上线 → 批量投递pending事件
  deliverDeferredEvents(deviceId: string): void {
    const queue: DistributedEvent[] | undefined =
      this.pendingEvents.get(deviceId);
    if (queue === undefined || queue.length === 0) return;

    console.info('[EventBus] 投递' + String(queue.length) +
      '个延迟事件到' + deviceId);

    // 按顺序投递
    for (let i: number = 0; i < queue.length; i++) {
      // 重新发布事件
      this.retryPublish(queue[i]);
    }

    this.pendingEvents.delete(deviceId);
  }
}

四、避坑速查

五、总结

分布式事件总线设计要点：

1. 全局唯一eventId + sequenceNumber保证顺序
2. TTL自动过期，防止KVStore膨胀
3. 延迟投递支持离线设备
4. 优先级队列确保关键事件优先处理