事件携带状态传输

概念定义

通知型事件只说“发生了什么”，消费者要回去问细节；状态传输型事件自带细节。前者事件小但耦合运行时，后者解耦运行时但付出载荷与副本存储。

工作机制

生产方按实体 id 与版本发布快照或增量；消费方 upsert 本地表并在本地连接查询。丢事件即副本陈旧：消费方跟踪序号并请求追赶快照或压缩日志重放。GDPR 式删除需要同等纪律的墓碑事件。

使用场景

多服务读同一份参考数据、消费方要低延迟读而不想跨服务调用、团队要独立部署与演进读模型的场景。载荷超过 broker 上限或高频变更导致副本抖动时不宜使用。

常见误区

发布了状态事件、消费方却仍回调取数，等于把刚拆掉的耦合重新装上。大集合塞进实体事件会把 MB 级消息灌进 broker。本地副本是收敛的读模型而非权威——业务决策写进副本就是写错地方。副本漂移是静默的：必须监控版本滞后；事件契约要版本化并做 upcast；敏感字段会随事件流遍各处，字段级治理要在契约评审里做。

代码示例

export class EventEmitter {
  #listeners = new Map();

  on(event, listener) {
    if (typeof event !== "string" || event.length === 0) {
      throw new TypeError("event must be a non-empty string");
    }
    if (typeof listener !== "function") {
      throw new TypeError("listener must be a function");
    }
    let bucket = this.#listeners.get(event);
    if (!bucket) {
      bucket = new Set();
      this.#listeners.set(event, bucket);
    }
    bucket.add(listener);
    return () => this.off(event, listener);
  }

  once(event, listener) {
    const wrapper = (...args) => {
      this.off(event, wrapper);
      listener(...args);
    };
    wrapper.original = listener;
    return this.on(event, wrapper);
  }

  off(event, listener) {
    const bucket = this.#listeners.get(event);
    if (!bucket) return false;
    let removed = false;
    for (const entry of bucket) {
      if (entry === listener || entry.original === listener) {
        bucket.delete(entry);
        removed = true;
      }
    }
    if (bucket.size === 0) this.#listeners.delete(event);
    return removed;
  }

  emit(event, ...args) {
    const bucket = this.#listeners.get(event);
    if (!bucket) return false;
    for (const listener of [...bucket]) listener(...args);
    return true;
  }

  listenerCount(event) {
    return this.#listeners.get(event)?.size ?? 0;
  }

  removeAllListeners(event) {
    if (event === undefined) this.#listeners.clear();
    else this.#listeners.delete(event);
  }
}

import { Kafka, logLevel } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'orders-service',
  brokers: (process.env.KAFKA_BROKERS ?? 'localhost:9092').split(','),
  logLevel: logLevel.WARN,
});

const producer = kafka.producer({ idempotent: true });
await producer.connect();

await producer.send({
  topic: 'orders.created',
  messages: [
    { key: 'customer-42', value: JSON.stringify({ orderId: 'ord-1', total: 99 }) },
    { key: 'customer-42', value: JSON.stringify({ orderId: 'ord-2', total: 12 }) },
  ],
});

const consumer = kafka.consumer({ groupId: 'billing-group' });
await consumer.connect();
await consumer.subscribe({ topic: 'orders.created', fromBeginning: false });

await consumer.run({
  autoCommit: false,
  eachBatchAutoResolve: false,
  eachMessage: async ({ topic, partition, message, heartbeat }) => {
    const order = JSON.parse(message.value?.toString() ?? '{}');

    console.log('billing for', order.orderId,
      'partition', partition, 'offset', message.offset);

    await heartbeat();
    await consumer.commitOffsets([
      { topic, partition, offset: (Number(message.offset) + 1).toString() },
    ]);
  },
});

process.on('SIGTERM', async () => {
  await consumer.disconnect();
  await producer.disconnect();
});