基于 Kafka 的医嘱事件架构设计

常见事件：OrderCreated、OrderUpdated、OrderCancelled、ExecutionReported、ChargePosted、ChargeReversed、EncounterOpened、EncounterClosed

五、Topic 命名与 Partition 策略

• 命名规范：<env>.<domain>.<entity>.<event>，例如 prod.his.order.created
• Partition key：使用 encounterId（或 orderId），保证相同住院/医嘱事件到同一 partition，从而保证顺序
• Partition 数：初期 8–24（根据并发消费者数），生产环境按吞吐与消费者扩展
• Retention 策略：
  • 事件 topic：保留 7–30 天（用于回放/补录）
  • 状态 topic（compacted）：prod.his.order.state（compact，保留最新状态）
  • 审计/归档 topic：长保留或导出到对象存储（S3 / HDFS）

六、事务一致性：Outbox 模式

为什么用 outbox？

直接在应用写 DB 后再发 Kafka 存在'写 DB 成功但发 Kafka 失败'的风险。outbox 把'写业务表 + 写 outbox 表'放在同一 DB 事务内，Debezium 将 outbox 的变更转成 Kafka 事件，从而实现'事务内写入 -> 事件最终到 Kafka'的原子性。

outbox 表示例 DDL（简化）

CREATE TABLE outbox (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  aggregate_id VARCHAR(64), -- e.g. encounterId
  event_type VARCHAR(64),
  payload TEXT,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  processed BOOLEAN DEFAULT FALSE
);

流程：

1. 业务事务写业务表 + 插入 outbox 行（同事务 commit）
2. Debezium/Connector 捕获 outbox 表增量 -> 写到 Kafka topic（并可删除或标记 processed）
3. 下游消费者处理 Kafka 事件（幂等）

七、Producer / Consumer 实践（示例代码片段）

下面给出 Java Spring Boot + Spring Kafka 的关键实现要点（伪代码，去掉细节配置）：

Consumer（幂等写 DB）

@KafkaListener(topics = "prod.his.order.created", groupId = "billing")
public void onOrderCreated(ConsumerRecord<String, OrderEvent> record) {
    OrderEvent event = record.value();
    String eventId = event.getEventId();
    if (processedEventRepository.exists(eventId)) { // 幂等：已经处理，直接返回
        return;
    }
    // 本地 DB 事务：写业务表 + processed_event 表
    txTemplate.execute(status -> {
        billingService.applyOrder(event.getPayload());
        processedEventRepository.insert(eventId, now());
        return null;
    });
}

关键点：

• 在本地 DB 事务内写入 processed_event(eventId)，保证幂等检查与写入原子性
• 若需要发出衍生事件（例如 ChargePosted），建议写入本地 outbox，再由 Debezium 推送，避免跨系统事务难题

Producer（若直接用 producer）

启用幂等性与事务（若不使用 outbox）：

enable.idempotence=true
transactional.id=his-producer-1

kafkaTemplate.executeInTransaction(kt -> {
    kt.send(topic, key, payload); // 本地 DB 写放在不同事务 -> 风险较大，不推荐此方式
    return true;
});

注意：直接 producer + 本地 DB 难以保证原子性，推荐 outbox + Debezium

八、CDC（Debezium / Kafka Connect）配置要点

• 使用 Debezium Connector 捕获 outbox 或业务表增量
• Debezium 写入 Kafka 时可设置转换（Kafka Connect SMT）把表 change event 转成领域事件（remove metadata，rename fields）
• 性能注意：不要直接把高写表全部 CDC 到 Kafka，否则会产生高流量；只 CDC 关键表或 outbox

示例 SMT：io.debezium.transforms.ExtractNewRecordState 将 payload.after 提取为 message value

九、流处理：Kafka Streams / ksqlDB 用例

常见场景：

• 将 OrderCreated 与 ExecutionReported join 形成 order.state（compacted）快照，供下游快速查询
• 实时聚合同一 encounter 的计费总额、执行次数，生成监控报警

ksql 伪示例：

CREATE STREAM order_created (...) WITH (kafka_topic='prod.his.order.created', value_format='AVRO');
CREATE STREAM exec_reported (...) WITH (kafka_topic='prod.his.execution.reported', value_format='AVRO');
CREATE TABLE order_state AS SELECT encounterId, LATEST_BY_OFFSET(payload) AS last_payload FROM order_created GROUP BY encounterId;

十、监控与报警

必监控项（分层）：

• Kafka Broker：UnderReplicatedPartitions, OfflinePartitionsCount, RequestHandlerAvgIdlePercent, DiskUsage
• Topic / Consumer：consumer_lag（每 consumer group）、throughput (MB/s)、partition skew
• 业务指标：事件延迟（produce->consume P95/P99）、DLQ 事件速率、事件处理失败率
• Alert 建议：consumer_lag > threshold、DLQ_rate > 0.5%、under_replicated_partitions > 0

监控栈：Prometheus + Grafana + JMX Exporter + (Confluent Control Center / Kafka Manager)

十一、安全与合规

• 传输加密：TLS（Kafka 集群 + clients）
• 认证与授权：SASL（Kerberos 或 SCRAM）+ ACL（topic/consumer group 级权限）
• 敏感数据处理：事件中尽量只传 patientId，PII 留在受控 DB；若必须携带敏感字段，使用字段级加密或脱敏
• 审计：记录 produce/consume 操作日志、DLQ 人工修复记录、谁触发了补偿事件等

十二、错误处理与补偿策略

1. 即时重试：消费失败短期内指数退避重试（供瞬时故障恢复）
2. Retry topic：将失败事件路由到延时 retry topic（例如 retry-1m, retry-5m）再尝试
3. DLQ：超次数失败进入 DLQ 并触发人工告警、监控看板与修复 UI（查看原始 payload、日志、手动修复/重放）
4. 补偿事件：当需要回滚（例如记账出错）发布 ChargeReversal / OrderCompensated 事件。补偿必须也是幂等的并写入审计

十三、性能与容量建议

• replication.factor = 3
• 每 broker 支持上千分区（视硬件），但单 topic 分区不宜过多以便运维
• 高流量 topic 初始 8–24 partitions（按消费并发再调）
• 估算：单事件大小 1–4 KB，2000 events/s ≈ 2–8 MB/s（HTTP + Kafka 双向），留足 headroom

十四、常见坑与对策（实战经验）

• 坑：事件丢失（应用写 DB 成功，但事件未进 Kafka）
  • 对策：Outbox + Debezium，保证原子性
• 坑：消费乱序导致状态错误
  • 对策：partition key 设计为 encounterId；使用 compacted order.state topic 保存最新状态
• 坑：schema 变动导致消费者宕机
  • 对策：严格 Schema Registry + 兼容策略（backward/forward）+ 版本化升级流程
• 坑：DLQ 积压无人处理
  • 对策：建立运维流程与 UI，自动告警与负责人制度
• 坑：敏感数据泄露
  • 对策：事件中不携带敏感 PII；必要时使用字段加密与访问控制

十五、落地实施步骤

1. PoC 环境：搭 3 节点 Kafka + Schema Registry + Debezium，接 test HIS DB
2. 实现 Outbox：在 HIS 事务中写 outbox；配置 Debezium 捕获并写入 prod.his.order.created
3. 实现 Consumer 幂等：processed_event 表 + 本地事务处理
4. 实现 DLQ & 修复 UI：运维可查看并手动重放/修正事件
5. 增加流处理：Kafka Streams/ksqlDB 生成 order.state compacted topic
6. 灰度：先在小科室跑，观察延迟、DLQ、处理率，再逐步全院推广

十六、示例附录

A. 简化 Avro schema（OrderCreated）

{
  "namespace": "hospital.his",
  "type": "record",
  "name": "OrderCreated",
  "fields": [
    {"name":"eventId","type":"string"},
    {"name":"eventType","type":"string"},
    {"name":"schemaVersion","type":"string"},
    {"name":"occurredAt","type":"string"},
    {"name":"source","type":"string"},
    {"name":"patientId","type":"string"},
    {"name":"encounterId","type":"string"},
    {"name":"orderId","type":"string"},
    {"name":"payload","type":{"type":"record","name":"OrderPayload","fields":[
      {"name":"items","type":{"type":"array","items":{"type":"record","name":"Item","fields":[
        {"name":"itemCode","type":"string"},
        {"name":"type","type":"string"},
        {"name":"qty","type":"int"},
        {"name":"unit","type":"string"}
      ]}}},
      {"name":"orderingPhysician","type":["null","string"], "default": null},
      {"name":"orderType","type":"string"}
    ]}}
  ]
}

B. Debezium outbox SMT 配置（示例片段）

{
  "name": "outbox-connector",
  "config": {
    "connector.class" : "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max" : "1",
    "database.hostname" : "db-host",
    "database.port" : "3306",
    "database.user" : "debezium",
    "database.password" : "pwd",
    "database.server.id" : "184054",
    "database.server.name" : "dbserver1",
    "table.include.list": "hisdb.outbox",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
  }
}

总结

• 核心建议：使用 Outbox + Debezium 保证事务一致性；以 encounterId 为 partition key 保证就诊内顺序；使用 Schema Registry 严格管理 schema；实现幂等消费与 DLQ+ 人工补偿机制。
• 事件化不是把一切改成事件就完成，而是要在主数据治理、监控、补偿链路上花功夫，才能真正把 HIS 的医嘱/费用流稳定地流转到病案、计费、DRG 与外部上报系统。