Event Driven

是什么

这是一份事件驱动架构规范，覆盖 Kafka（分布式消息队列）分区策略、outbox（发件箱）模式、死信队列、消费者幂等、事件溯源等核心模式，让系统从同步调用转成异步事件流时不掉消息、不脏数据。

怎么用

1. 设计新事件流时，先按本文档判断属于命令、领域事件还是集成事件，再选发布渠道。
2. 数据库写入和消息发布要原子时套用 outbox 模式，避免双写不一致导致下游漏单。
3. 消费者侧实现处理逻辑时按幂等模板编码，重复消费同一条消息不会造成业务后果二次发生。
4. 失败消息进入 DLQ（死信队列）后按文档定义的处置流程介入，定期清理或回放，避免堆积变成黑洞。
5. 上线前过一遍分区数、副本数、保留期、监控四项配置，按业务峰值留出至少 2 倍冗余。

架构图

flowchart LR
    A[业务服务] --> B[Outbox 表]
    B --> C[Kafka 主题]
    C --> D[消费者集群]
    D --> E{处理成功?}
    E -->|否| F[死信队列]

Event-Driven Architecture Patterns

Design and implementation patterns for reliable, scalable asynchronous systems using message brokers.

When to Activate

• Decoupling services that currently call each other synchronously
• Designing Kafka topics, partitions, or consumer group topology
• Implementing event schemas with versioning and backward compatibility
• Ensuring exactly-once or at-least-once delivery semantics
• Setting up dead-letter queues and poison message handling
• Implementing the outbox pattern to avoid dual-write problems
• Building event sourcing or CQRS read-model projections

Core Concepts

Broker Comparison

Delivery Semantics

Default to at-least-once with idempotent consumers — it's simpler than exactly-once and eliminates loss.

Event Schema Design

Event Envelope

{
  "event_id": "01HXK4...",
  "event_type": "order.placed",
  "schema_version": "1.0",
  "occurred_at": "2025-06-01T12:00:00Z",
  "aggregate_id": "order-abc-123",
  "aggregate_type": "Order",
  "correlation_id": "req-xyz-456",
  "payload": {
    "customer_id": "cust-789",
    "total": "59.99",
    "currency": "USD",
    "items": [{ "sku": "WIDGET-01", "qty": 2, "unit_price": "29.99" }]
  }
}

Rules:

• event_id: globally unique (ULID or UUID v7) — used for idempotency keys
• occurred_at: when the fact happened, not when published
• correlation_id: propagate from the originating request for distributed tracing
• Monetary values: string decimal, not float

Schema Versioning (Avro / JSON Schema)

# GOOD: backward-compatible change — add optional field with default
{
  "event_type": "order.placed",
  "schema_version": "1.1",   # minor bump — consumers on 1.0 still work
  "payload": {
    "customer_id": "...",
    "total": "...",
    "promo_code": null        # new optional field, defaults to null
  }
}

# BAD: breaking changes
# - removing a field consumers depend on
# - renaming a field
# - changing a field type (string → int)
# → always create a new event type or bump major version

Kafka Patterns

Topic and Partition Design

# Naming: <domain>.<entity>.<event-type> or <domain>.<entity>
orders.placed
orders.cancelled
payments
inventory.stock-updates

# Partition key selection — determines ordering guarantee
producer.produce(topic="orders", key=order.customer_id, ...)  # all orders per customer are ordered
producer.produce(topic="payments", key=payment.order_id, ...)  # all events per order are ordered

# Partition count tradeoffs
low partitions (1–6):   easy rebalancing, less parallelism
high partitions (12+):  more parallelism, more consumer instances, slower rebalance
rule: start with max(consumer_instances * 2, 12), increase later

Python Producer

from confluent_kafka import Producer, KafkaException
import json, uuid
from datetime import datetime, timezone

producer = Producer({
    "bootstrap.servers": "kafka:9092",
    "acks": "all",                    # wait for all ISR replicas
    "retries": 5,
    "retry.backoff.ms": 500,
    "enable.idempotence": True,       # exactly-once producer semantics
    "compression.type": "lz4",
})

def publish_event(topic: str, key: str, event_type: str, payload: dict):
    event = {
        "event_id": str(uuid.uuid4()),
        "event_type": event_type,
        "schema_version": "1.0",
        "occurred_at": datetime.now(timezone.utc).isoformat(),
        "payload": payload,
    }
    producer.produce(
        topic=topic,
        key=key.encode(),
        value=json.dumps(event).encode(),
       
    )
    producer.poll(0)  # trigger callbacks without blocking

def _delivery_report(err, msg):
    if err:
        logger.error("delivery_failed", topic=msg.topic(), error=str(err))

producer.flush()  # call at shutdown

Python Consumer

from confluent_kafka import Consumer, KafkaError
import json

consumer = Consumer({
    "bootstrap.servers": "kafka:9092",
    "group.id": "notification-service",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,      # manual commit after processing
    "max.poll.interval.ms": 300_000,
})

consumer.subscribe(["orders"])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError.PARTITION_EOF:
                continue
            raise KafkaException(msg.error())

        event = json.loads(msg.value())
        try:
            handle_event(event)
            consumer.commit(message=msg)   # commit only after success
        except Exception as e:
            logger.error("processing_failed", event_id=event["event_id"], error=str(e))
            # don't commit — message will be redelivered
finally:
    consumer.close()

TypeScript Consumer (KafkaJS)

import { Kafka } from "kafkajs";

const kafka = new Kafka({ brokers: ["kafka:9092"] });
const consumer = kafka.consumer({ groupId: "notification-service" });

await consumer.connect();
await consumer.subscribe({ topics: ["orders"], fromBeginning: false });

await consumer.run({
  autoCommit: false,
  eachMessage: async ({ topic, partition, message, heartbeat }) => {
    const event = JSON.parse(message.value!.toString());
    try {
      await handleEvent(event);
      await consumer.commitOffsets([{
        topic, partition,
        offset: (Number(message.offset) + 1).toString(),
      }]);
    } catch (err) {
      logger.error({ eventId: event.event_id, err }, "processing_failed");
      throw err;  // re-throw to pause and retry
    }
  },
});

Go Consumer (Sarama)

import "github.com/IBM/sarama"

type OrderHandler struct{ db *sql.DB }

func (h *OrderHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (h *OrderHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

func (h *OrderHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        var event Event
        if err := json.Unmarshal(msg.Value, &event); err != nil {
            logger.Error("unmarshal_failed", "offset", msg.Offset)
            sess.MarkMessage(msg, "")  // skip unparseable messages
            continue
        }
        if err := h.handle(sess.Context(), event); err != nil {
            return err  // stop processing; session will restart
        }
        sess.MarkMessage(msg, "")
    }
    return nil
}

Idempotency

Consumers must handle duplicate messages — at-least-once delivery guarantees redelivery on failure.

# Pattern: idempotency key in a processed-events table
def handle_order_placed(event: dict, db: Session):
    event_id = event["event_id"]

    # Check if already processed
    if db.query(ProcessedEvent).filter_by(event_id=event_id).first():
        logger.info("duplicate_skipped", event_id=event_id)
        return

    # Process in a transaction that also inserts the idempotency record
    with db.begin():
        create_notification(event["payload"])
        db.add(ProcessedEvent(event_id=event_id, processed_at=datetime.utcnow()))

// Alternative: upsert on natural key (e.g. order_id)
await db.transaction(async (trx) => {
  await trx("notifications")
    .insert({ order_id: event.payload.order_id, customer_id: event.payload.customer_id })
    .onConflict("order_id")
    .ignore();  // safe to call multiple times
});

Outbox Pattern

Solves dual-write: never publish directly from application code after a DB write — they can desync.

# BAD: dual write (either step can fail independently)
BEGIN TRANSACTION
  INSERT INTO orders (...)
COMMIT
kafka.produce("order.placed", ...)   # ← this can fail silently

# GOOD: outbox pattern
BEGIN TRANSACTION
  INSERT INTO orders (...)
  INSERT INTO outbox (event_type, payload, published=false, ...)  # same transaction
COMMIT

# Separate relay process (Debezium CDC or polling relay)
SELECT * FROM outbox WHERE published = false ORDER BY created_at LIMIT 100
→ produce to Kafka
→ UPDATE outbox SET published = true WHERE id IN (...)

# Polling relay (simple, no CDC dependency)
def relay_outbox(db: Session, producer: Producer):
    events = db.query(OutboxEvent).filter_by(published=False).limit(100).all()
    for event in events:
        producer.produce(
            topic=event.topic,
            key=event.aggregate_id.encode(),
            value=event.payload.encode(),
        )
    producer.flush()
    for event in events:
        event.published = True
    db.commit()

Dead-Letter Queues

# Python — send to DLQ after max retries
MAX_RETRIES = 3

def handle_with_dlq(consumer, dlq_producer, msg):
    retry_count = int(msg.headers().get("retry-count", b"0"))
    event = json.loads(msg.value())

    try:
        process_event(event)
        consumer.commit(message=msg)
    except Exception as e:
        if retry_count >= MAX_RETRIES:
            dlq_producer.produce(
                topic=f"{msg.topic()}.dlq",
                key=msg.key(),
                value=msg.value(),
                headers={"original-topic": msg.topic(), "error": str(e)},
            )
            consumer.commit(message=msg)  # move past it
            logger.error("sent_to_dlq", event_id=event["event_id"])
        else:
            # Re-queue with incremented retry count (or let Kafka retry via pause)
            logger.warning("retry", attempt=retry_count + 1, event_id=event["event_id"])

DLQ naming convention: <original-topic>.dlq DLQ review: alert on DLQ lag > 0; investigate and replay or discard manually.

Event Sourcing

Store state as an append-only log of events; derive current state by replaying.

# Events are facts, not commands
EVENTS = [
    {"type": "OrderCreated",   "payload": {"customer_id": "c1", "items": [...]}},
    {"type": "ItemAdded",      "payload": {"sku": "X", "qty": 1}},
    {"type": "OrderConfirmed", "payload": {"confirmed_at": "2025-01-01T..."}},
]

def replay(events: list[dict]) -> Order:
    order = Order()
    for event in events:
        match event["type"]:
            case "OrderCreated":   order.apply_created(event["payload"])
            case "ItemAdded":      order.apply_item_added(event["payload"])
            case "OrderConfirmed": order.apply_confirmed(event["payload"])
    return order

# Snapshot: periodically persist current state to avoid full replay
# Projection: consume event stream to build read models (CQRS read side)

See also: microservices, observability

Red Flags

• Kafka consumer without idempotency — at-least-once delivery means the same message can arrive twice; design consumers to be idempotent before assuming exactly-once semantics
• Hot partition from a low-cardinality key — using event_type as the partition key routes all messages of one type to one partition; choose a high-cardinality key like entity_id for even distribution
• No dead-letter queue for unprocessable messages — a consumer that throws on a bad message blocks all subsequent messages on that partition; route poison messages to a DLQ immediately
• Outbox table without a reliable poller — writing to the outbox without a dedicated transactional poller means events may silently never be published; the poller is half the pattern
• Schema changes without versioning — adding a required field to an event schema breaks all existing consumers silently; always version events and maintain backward compatibility
• Synchronous HTTP calls inside a consumer handler — an upstream timeout blocks the consumer and grows partition lag; use async clients or pre-fetch data outside the consumer loop
• Resetting offsets to earliest on every consumer restart — without committed offsets, a restarted consumer reprocesses all historical events; commit offsets after processing and handle replay explicitly

Checklist

• Event envelope includes event_id, event_type, schema_version, occurred_at, correlation_id
• Partition key chosen to guarantee ordering for events that must be ordered
• Producer uses acks=all and enable.idempotence=true
• Consumer commits offsets manually after successful processing
• All consumers are idempotent (duplicate event_id handled gracefully)
• Outbox pattern used — never dual-write to DB and broker in separate transactions
• Dead-letter queue configured; alerts fire when DLQ lag exceeds threshold
• Event schema changes are backward-compatible (add optional fields only)
• Consumer group IDs are service-specific and stable across deploys
• Dead-letter messages include original topic, offset, and error reason in headers
• Retention period set to cover the longest plausible consumer downtime plus buffer