訊息佇列與事件驅動

為什麼需要訊息佇列，以及它的代價 基礎 高頻

訊息佇列（Message Queue, MQ）在生產者與消費者之間放一個可靠的緩衝區。面試時被問「為什麼要用 MQ」，要能講出四個好處，並且主動講代價——這是區分「用過」與「懂」的關鍵。

代價：引入 MQ 會帶來什麼問題

• 系統複雜度與維運成本：多了一個需要高可用、監控、容量規劃、升級的分散式系統。
• 最終一致性：寫入後下游不會立即看到；使用者可能在「訂單建立」後立刻查詢卻還看不到積分。
• 交付語意問題：訊息可能重複（at-least-once）或遺失（at-most-once），消費者必須冪等。
• 順序問題：多 partition / 多消費者 / 重試都可能打亂順序。
• 除錯困難：請求的因果鏈被切斷，需要 correlation id 與分散式追蹤；「訊息去哪了」是常見的疑難雜症。
• Dual write：同時寫 DB 與 MQ 無法原子化，需要 outbox pattern。
• 延遲：多一跳，加上批次等待，端到端延遲從毫秒變成數十毫秒到秒級。

Point-to-Point vs Publish/Subscribe

Kafka 同時支援兩者：同一個 consumer group 內是競爭消費（每個 partition 只給組內一個 consumer），不同 consumer group 之間是 pub/sub（每個 group 各自讀完整份資料、各自維護 offset）。

傳統佇列 vs Log-based 訊息系統

Kafka 核心概念：Topic、Partition、Offset、Segment 進階 高頻

Kafka 本質上是一個分散式、分區、多副本的 append-only commit log。理解它的關鍵是把它想成「一堆可以被多方獨立讀取的檔案」，而不是「佇列」。

Topic

邏輯上的訊息分類（如 impressions、clicks）。一個 topic 由多個 partition 組成。

Partition

Kafka 平行度、順序性、儲存的基本單位。每個 partition 是一個有序、不可變的訊息序列，只能在尾端 append。順序只在 partition 內保證，跨 partition 沒有全域順序。partition 數可以增加但不能減少，且增加會改變 key → partition 的對應。

Offset

訊息在 partition 中的序號（64-bit 單調遞增）。consumer 以 (topic, partition, offset) 定位讀取位置。commit offset 代表「下一則要讀的位置」。

Broker

Kafka 伺服器節點。每個 partition 的多個副本分散在不同 broker 上，其中一個是 leader。

Segment

partition 在磁碟上被切成多個 segment 檔（預設 log.segment.bytes = 1GB，或依時間 roll）。每個 segment 有 .log（資料）、.index（offset → 檔案位置的稀疏索引）、.timeindex（時間戳 → offset）。只有最新的 active segment 會被寫入；retention 與 compaction 以 segment 為單位刪除或清理，所以刪除舊資料就是刪檔案，非常便宜。

Record

key（可為 null）、value、timestamp、headers。多筆 record 組成 record batch，batch 是壓縮與傳輸的單位。

# partition 在磁碟上的樣子
$ ls /var/lib/kafka/data/ad-events-0/
00000000000000000000.index      00000000000000000000.log      00000000000000000000.timeindex
00000000000005367851.index      00000000000005367851.log      00000000000005367851.timeindex
leader-epoch-checkpoint         partition.metadata
# 檔名 = 該 segment 第一則訊息的 offset；找 offset 5400000：二分搜尋檔名 → 查 .index 稀疏索引 → 從該位置循序掃描

# 建立 topic
kafka-topics.sh --bootstrap-server kafka:9092 --create --topic ad-events \
  --partitions 48 --replication-factor 3 \
  --config min.insync.replicas=2 --config retention.ms=259200000 --config compression.type=zstd

Retention 與 Log Compaction

也可以 compact,delete 同時使用：compact 之外仍刪除過舊資料。Compacted topic 讓新服務可以「從頭讀一遍」就重建出每個 key 的最新狀態，是 event-carried state transfer 與 Kafka Streams KTable 的基礎。

KRaft：取代 ZooKeeper

舊版 Kafka 用 ZooKeeper 存放 metadata（broker 列表、topic 設定、partition leader 與 ISR）並選出單一 controller。問題：要維運兩套分散式系統；controller failover 時要從 ZooKeeper 重新載入所有 metadata，partition 數多（數十萬）時非常慢；metadata 傳播不一致。

KRaft（KIP-500）把 metadata 管理移進 Kafka 本身：一組 controller 節點以 Raft 協定維護一個內部的 metadata log（__cluster_metadata），其中一個是 active controller；broker 像 consumer 一樣訂閱這個 log 取得 metadata。好處：單一系統、controller failover 幾乎即時（standby controller 已有完整 metadata）、可支援更多 partition。KRaft 在 3.3 宣告可用於正式環境，Kafka 4.0（2025）完全移除 ZooKeeper 模式。

Kafka 複製：Leader/Follower、ISR、acks 深入 高頻

每個 partition 有 replication.factor 個副本（通常 3），分散在不同 broker（最好不同機架/AZ，透過 broker.rack 設定）。其中一個是 leader，所有讀寫預設都走 leader（2.4 之後 consumer 可以從同機架的 follower 讀，以節省跨 AZ 流量）；follower 像 consumer 一樣持續從 leader 拉資料。

ISR（In-Sync Replicas）、HW 與 LEO

LEO（Log End Offset）

每個副本自己 log 的下一個寫入位置。

ISR

與 leader「保持同步」的副本集合（含 leader 自己）。follower 若超過 replica.lag.time.max.ms（目前預設 30 秒）沒有追上 leader 的 LEO，就會被移出 ISR；追上後再加回來。

HW（High Watermark）

ISR 中所有副本都已擁有的最大 offset。只有 HW 以下的訊息（committed）才對 consumer 可見，避免 consumer 讀到之後可能因 leader 切換而消失的訊息。

Leader epoch

每次換 leader 遞增，用於副本恢復時截斷不一致的 log（取代早期只依 HW 截斷造成的資料遺失/分歧問題），概念上就是 Raft 的 term。

acks 與 min.insync.replicas

重點：acks=all 等的是「目前的 ISR」，如果 ISR 縮到只剩 leader 一個，acks=all 就退化成 acks=1。所以要設 min.insync.replicas（topic 或 broker 層級，預設 1）：ISR 數量少於此值時，leader 拒絕 acks=all 的寫入（回 NotEnoughReplicas），寧可不可用也不冒遺失風險。

Unclean Leader Election

如果 leader 掛了，而 ISR 中沒有任何其他存活副本（例如所有 follower 早就落後被踢出 ISR），該怎麼辦？

• unclean.leader.election.enable=false（預設，自 0.11 起）：等待某個 ISR 副本恢復，期間該 partition 不可用。選一致性（C）。
• true：允許非 ISR 的落後副本當 leader，立即恢復可用，但它缺少的訊息永久遺失，且之後舊 leader 回來時會截斷自己的 log 以對齊。選可用性（A）。

這是 CAP 取捨在 Kafka 中的具體呈現：計費、交易類 topic 保持 false；可容忍少量遺失的 metrics/log topic 可以考慮 true。

Kafka Producer：批次、壓縮、冪等、Partitioner 與順序 進階 高頻

Producer 的 send() 是非同步的：訊息先放進記憶體中依 partition 分組的 buffer（RecordAccumulator），由背景 sender thread 批次送出。

Batching、linger.ms、Compression

壓縮的 batch 在 broker 端不解壓直接存（若 topic 的 compression.type 與 producer 相同或為 producer），consumer 收到後才解壓——端到端壓縮同時省網路、磁碟與 broker CPU。所以 batch 越大，壓縮比越好，吞吐越高。

Idempotent Producer

問題：producer 送出 batch，broker 已寫入但 ack 在網路上遺失，producer 重試 → 重複訊息。另外 in-flight > 1 時，batch 1 失敗重試、batch 2 成功，會造成亂序。

enable.idempotence=true：broker 為每個 producer 分配 Producer ID（PID），producer 為每個 partition 的每個 batch 加上遞增的 sequence number。broker 記住每個 (PID, partition) 最近的 sequence：重複的直接丟棄並回成功、跳號的拒絕。效果：單一 producer session 對單一 partition 的寫入不重複、不亂序（允許 in-flight ≤ 5）。限制：producer 重啟會拿到新 PID，跨 session 的重複無法消除——這需要 transactions 的 transactional.id。

Partitioner、Key 選擇與順序保證

• 有 key：預設以 murmur2(key) % partition 數決定 partition。同 key → 同 partition → 有序。
• 無 key（null）：2.4 起用 sticky partitioner：先把訊息黏在一個 partition 直到 batch 送出再換，讓 batch 更大（舊版 round-robin 每則換 partition 會讓 batch 很小）。
• 也可以自訂 partitioner，或直接指定 partition。

Python Producer 範例（confluent-kafka）

import json
import socket
import uuid
from confluent_kafka import Producer, KafkaException

producer = Producer({
    "bootstrap.servers": "kafka-1:9092,kafka-2:9092,kafka-3:9092",
    "client.id": socket.gethostname(),
    "acks": "all",                    # 等所有 ISR 確認
    "enable.idempotence": True,       # PID + sequence，避免重試造成的重複與亂序
    "compression.type": "zstd",
    "linger.ms": 20,                  # 最多等 20ms 湊 batch
    "batch.size": 256 * 1024,
    "delivery.timeout.ms": 120_000,
    "queue.buffering.max.kbytes": 512 * 1024,   # librdkafka 本地 buffer 上限
})

def on_delivery(err, msg):
    # 在 poll()/flush() 呼叫時於呼叫端 thread 執行；不要在這裡做重活
    if err is not None:
        # 已經超過 delivery.timeout.ms 的重試仍失敗 → 寫入本地備援（磁碟/DLQ）並告警
        print(f"delivery failed key={msg.key()} err={err}")

def send_event(event: dict) -> None:
    event.setdefault("event_id", str(uuid.uuid4()))          # 給下游去重用
    while True:
        try:
            producer.produce(
                topic="ad-events",
                key=event["user_id"].encode(),                # 同 user 同 partition → 有序
                value=json.dumps(event, separators=(",", ":")).encode(),
                headers={"schema": b"ad_event.v3"},
                on_delivery=on_delivery,
            )
            break
        except BufferError:
            producer.poll(0.1)        # 本地 buffer 滿：處理 callback 讓空間釋放（backpressure）
    producer.poll(0)                  # 觸發已完成的 delivery callback

# 程式結束或 SIGTERM 時務必 flush，否則 buffer 中的訊息會遺失
def shutdown():
    remaining = producer.flush(10)
    if remaining:
        print(f"{remaining} messages not delivered")

Kafka Consumer：Consumer Group、Rebalance、Offset Commit、Lag 深入 高頻

Consumer Group

同一個 group.id 的 consumer 組成一個 group，topic 的 partition 被分配給組內成員，每個 partition 同一時間只屬於組內一個 consumer。推論：

• 組內 consumer 數 > partition 數時，多出來的 consumer 閒置。partition 數就是單一 group 的最大平行度。
• 一個 consumer 可以處理多個 partition。
• 每個 group 有一個 group coordinator（某個 broker），負責成員管理與 offset 儲存（寫入 __consumer_offsets）。

若需要比 partition 數更高的處理平行度（例如每則訊息要呼叫 200ms 的外部 API），可以在 consumer 內部用 worker pool 平行處理，但要自行追蹤「哪些 offset 已完成」，只 commit 連續完成的最小 offset（Confluent 的 Parallel Consumer 函式庫就是做這件事，並可按 key 保持順序）。

Rebalance：Eager vs Cooperative

成員加入、離開、被判定死亡（超過 session.timeout.ms 沒心跳，預設 45 秒；或兩次 poll() 間隔超過 max.poll.interval.ms，預設 5 分鐘），或 topic partition 數改變時，會觸發 rebalance 重新分配 partition。

減少不必要 rebalance 的手段：

• Static membership（group.instance.id，KIP-345）：給每個 consumer 固定身分（例如 StatefulSet 的 pod 名稱），短暫重啟在 session timeout 內回來就不觸發 rebalance。
• 單批處理時間要遠小於 max.poll.interval.ms：減少 max.poll.records（預設 500），或把慢的處理移到別的 thread 並用 pause/resume。
• 在 on_revoke callback 中 commit 已處理的 offset，避免 partition 移交後被新 owner 重複處理。

Offset Commit 策略與交付語意

auto.offset.reset：group 沒有已 commit 的 offset（新 group 或 offset 已過期被刪）時從哪開始：latest（預設，只讀新訊息）或 earliest（從頭讀）。新的關鍵消費者通常用 earliest，否則上線前的訊息會被跳過。

Python Consumer：手動 commit 的 at-least-once 範例

import json
import signal
from confluent_kafka import Consumer, KafkaError, KafkaException, TopicPartition

running = True
def _stop(*_):
    global running
    running = False
signal.signal(signal.SIGTERM, _stop)   # K8s 終止 pod 時先送 SIGTERM，給我們優雅關閉的機會

consumer = Consumer({
    "bootstrap.servers": "kafka-1:9092,kafka-2:9092",
    "group.id": "impression-aggregator",
    "group.instance.id": "impression-aggregator-0",       # static membership（例如用 pod 名）
    "enable.auto.commit": False,                           # 自己決定何時 commit
    "auto.offset.reset": "earliest",
    "partition.assignment.strategy": "cooperative-sticky",
    "max.poll.interval.ms": 300_000,
    "isolation.level": "read_committed",                   # 不讀未提交的交易訊息
})

def on_revoke(c, partitions):
    # partition 即將被移走：先 commit 已處理完的位置，降低新 owner 重複處理量
    try:
        c.commit(asynchronous=False)
    except KafkaException:
        pass

consumer.subscribe(["ad-events"], on_revoke=on_revoke)

def process_batch(msgs) -> None:
    rows = [json.loads(m.value()) for m in msgs]
    sink.upsert_many(rows)          # 冪等寫入（以 event_id 為唯一鍵），重複處理也安全

try:
    while running:
        msgs = consumer.consume(num_messages=500, timeout=1.0)   # 批次取，提升吞吐
        if not msgs:
            continue
        good = []
        for m in msgs:
            if m.error():
                if m.error().code() == KafkaError._PARTITION_EOF:
                    continue
                raise KafkaException(m.error())
            good.append(m)
        process_batch(good)                       # 先處理（寫入成功）……
        consumer.commit(asynchronous=False)       # ……再 commit：at-least-once
finally:
    consumer.close()    # 會離開 group 並觸發 rebalance（static member 則等 session timeout）

Consumer Lag 監控

Lag = partition 的 log end offset − group 已 commit 的 offset。它是 consumer 健康度最重要的指標。

$ kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group impression-aggregator
GROUP                  TOPIC      PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG     CONSUMER-ID
impression-aggregator  ad-events  0          91823311        91823950        639     impression-aggregator-0-...
impression-aggregator  ad-events  1          90011020        90950021        939001  impression-aggregator-1-...
# partition 1 的 lag 特別大 → 熱 partition（key skew）或該 consumer 卡住

# 重設 offset（group 必須沒有活躍成員）：修 bug 後重播 2 小時前的資料
$ kafka-consumer-groups.sh --bootstrap-server kafka:9092 --group impression-aggregator \
    --topic ad-events --reset-offsets --to-datetime 2026-11-03T10:00:00.000 --execute

• 以「時間」衡量 lag 比「訊息數」直觀：100 萬則 lag 在尖峰可能只是 2 秒，在離峰可能是 1 小時。可以用「目前處理訊息的 timestamp 與現在的差距」或 lag ÷ 消費速率估算。
• 告警看趨勢而非單點：lag 持續上升超過 N 分鐘、或 lag 時間超過業務 SLA（例如報表允許 5 分鐘延遲）。LinkedIn 的 Burrow 就是用滑動視窗評估 lag 趨勢與 consumer 是否停滯。
• 工具：kafka-consumer-groups.sh、Burrow、kafka-lag-exporter / kafka_exporter + Prometheus、雲端服務（MSK、Confluent Cloud）的內建指標。
• Lag 持續增加的處置：擴 consumer（不超過 partition 數）→ 優化處理（批次寫入、減少同步呼叫）→ 增加 partition → 檢查熱 partition 與 GC/外部依賴。

Kafka Exactly-once Semantics 與為什麼 Kafka 這麼快 深入

Exactly-once：Idempotent Producer + Transactions

Kafka 的 EOS（0.11 起）針對的是 consume-transform-produce 流程：從 topic A 讀、處理、寫到 topic B，並 commit A 的 offset。目標是「寫出 B 的結果」與「commit A 的 offset」原子地一起生效。

1. Producer 設定 transactional.id（跨重啟穩定的 id），呼叫 init_transactions()：transaction coordinator 依 transactional.id 分配 PID 並遞增 epoch，舊的同 id 實例（zombie）的寫入會被 fence 掉。
2. begin_transaction() → 送出多則訊息到多個 partition。
3. send_offsets_to_transaction(offsets, group_metadata)：把消費的 offset 也納入同一個交易。
4. commit_transaction()：coordinator 以類似 2PC 的方式在 transaction log 記錄決定，再對所有相關 partition 寫入 commit marker。失敗則 abort_transaction()。
5. 下游 consumer 設 isolation.level=read_committed，只讀到已 commit 交易的訊息（讀到 LSO，Last Stable Offset 為止），aborted 的訊息被過濾掉。

from confluent_kafka import Consumer, Producer

consumer = Consumer({"bootstrap.servers": BROKERS, "group.id": "enricher",
                     "enable.auto.commit": False, "isolation.level": "read_committed"})
producer = Producer({"bootstrap.servers": BROKERS, "transactional.id": "enricher-0"})
producer.init_transactions()
consumer.subscribe(["raw-clicks"])

while True:
    msgs = consumer.consume(500, timeout=1.0)
    if not msgs:
        continue
    producer.begin_transaction()
    try:
        for m in msgs:
            if m.error():
                continue
            producer.produce("enriched-clicks", key=m.key(), value=enrich(m.value()))
        # offset 與輸出在同一交易：要嘛都生效，要嘛都不生效
        producer.send_offsets_to_transaction(
            consumer.position(consumer.assignment()),
            consumer.consumer_group_metadata())
        producer.commit_transaction()
    except Exception:
        producer.abort_transaction()
        rewind_to_last_committed(consumer)   # seek 回上次 commit 的位置重新處理

為什麼 Kafka 這麼快

RabbitMQ 與 Kafka 比較 進階

RabbitMQ 是實作 AMQP 0-9-1 的傳統 message broker（Erlang 撰寫），核心是「智慧的 broker、單純的 consumer」：broker 負責路由、追蹤每則訊息的 ack、重新投遞。

模型：Producer → Exchange → Binding → Queue → Consumer

Producer 不直接寫 queue，而是寫到 exchange，exchange 依 binding 規則與訊息的 routing key 決定送到哪些 queue。

Ack/Nack、Prefetch、DLX

Consumer ack

手動 ack 模式下，consumer 處理完送 basic.ack，broker 才刪除訊息。若 consumer 的連線斷了而尚未 ack，訊息會重新投遞給其他 consumer（at-least-once）。basic.nack/basic.reject 可選擇 requeue=true（放回佇列）或 false（丟棄或進 DLX）。

Prefetch（basic.qos）

每個 consumer 最多可持有多少則未 ack 的訊息。沒設定（無上限）時 broker 會把大量訊息推給第一個 consumer，造成負載不均與記憶體暴增；設太小則吞吐低（每則都要等 RTT）。常見從 10–300 之間依處理耗時調整。這是 RabbitMQ 的 push 模型下的 backpressure 機制。

Publisher confirms

producer 端的 ack：broker 把訊息持久化（或複製到 quorum）後回 confirm。與 persistent message、durable queue 一起才能保證不遺失。

DLX（Dead Letter Exchange）

queue 設定 x-dead-letter-exchange，訊息在以下情況被轉送到 DLX：被 reject/nack 且 requeue=false、訊息 TTL 到期、queue 超過長度上限。結合 TTL 可以實作延遲重試：失敗訊息送到「等待 queue」（TTL 30 秒、無 consumer），到期後經 DLX 回到原 queue。

Queue 類型

Classic queue（單節點）、Quorum queue（以 Raft 複製，取代已被移除的 classic mirrored queue，是高可用的推薦選擇）、Stream（3.9 起，append-only log，可重播，類似 Kafka）。

Kafka vs RabbitMQ

其他訊息系統：SQS/SNS、Google Pub/Sub、Redis Streams、NATS、Pulsar 進階

# SQS 消費者：處理成功才 delete；失敗就不刪，visibility timeout 到期後自動重投
import boto3, json
sqs = boto3.client("sqs")
URL = "https://sqs.ap-northeast-1.amazonaws.com/123456789012/push-jobs"

while True:
    resp = sqs.receive_message(QueueUrl=URL, MaxNumberOfMessages=10,
                               WaitTimeSeconds=20,           # long polling，減少空請求費用
                               VisibilityTimeout=60)         # 需大於單則最長處理時間
    for msg in resp.get("Messages", []):
        try:
            handle(json.loads(msg["Body"]))                  # 必須冪等：同一則可能被投遞多次
            sqs.delete_message(QueueUrl=URL, ReceiptHandle=msg["ReceiptHandle"])
        except Exception:
            pass   # 不刪除 → 逾時後重投；超過 maxReceiveCount 由 redrive policy 送進 DLQ

# Redis Streams consumer group
import redis
r = redis.Redis()
try:
    r.xgroup_create("push:stream", "senders", id="0", mkstream=True)
except redis.ResponseError:
    pass                                                    # group 已存在

r.xadd("push:stream", {"user_id": "u42", "campaign": "c7"}, maxlen=1_000_000, approximate=True)

while True:
    # ">" 代表只拿從未投遞給本 group 的新訊息
    resp = r.xreadgroup("senders", "worker-1", {"push:stream": ">"}, count=100, block=5000)
    for _stream, entries in resp or []:
        for msg_id, fields in entries:
            send_push(fields)
            r.xack("push:stream", "senders", msg_id)
    # 定期接手其他 worker 掛掉後留在 PEL 中超過 60 秒的訊息
    r.xautoclaim("push:stream", "senders", "worker-1", min_idle_time=60_000, count=100)

交付語意、冪等消費者、順序、DLQ 與 Backpressure 深入 高頻

三種交付語意

冪等消費者設計

重複的來源：producer 重試、consumer 處理完但 commit 前當機、rebalance、SQS visibility timeout 到期、上游重送。對策依場景選擇：

1. 天然冪等的操作：UPSERT ... SET status='SENT'、以 event_id 為主鍵寫入。重複寫同一筆資料無害。
2. 去重表（dedup table）：processed_events(event_id PRIMARY KEY)，處理與插入去重紀錄在同一個交易中；插入衝突代表已處理過，直接跳過。
3. 快取去重：Redis SET event:{id} 1 NX EX 86400，NX 失敗代表處理過。較快但與業務寫入不在同一交易，處理失敗時要刪除 key（或接受極小機率的漏處理/重複）。
4. 版本號 / 序號：每個實體的事件帶遞增 version，consumer 只套用 version > 目前 version 的事件，同時解決重複與亂序（CDC 常用）。
5. Idempotency key 傳到外部系統：呼叫第三方（支付、LINE 推播）時帶上 idempotency/retry key，讓對方去重。

# 冪等消費者：去重紀錄與業務寫入在同一個 DB 交易中（PostgreSQL + psycopg 3）
import json

def handle_conversion(conn, msg) -> None:
    ev = json.loads(msg.value())
    with conn.transaction():
        cur = conn.execute(
            "INSERT INTO processed_events (consumer, event_id) VALUES (%s, %s) "
            "ON CONFLICT DO NOTHING",
            ("conversion-attributor", ev["event_id"]))
        if cur.rowcount == 0:
            return                                    # 已處理過：跳過（但仍允許 commit offset）
        conn.execute(
            "INSERT INTO campaign_daily_stats (campaign_id, day, conversions, revenue) "
            "VALUES (%s, %s, 1, %s) "
            "ON CONFLICT (campaign_id, day) DO UPDATE "
            "SET conversions = campaign_daily_stats.conversions + 1, "
            "    revenue = campaign_daily_stats.revenue + EXCLUDED.revenue",
            (ev["campaign_id"], ev["day"], ev["revenue"]))
    # 交易提交後才 commit Kafka offset；若此時當機，重投的訊息會被去重表擋下

# processed_events 需定期清理（保留時間 > 最大可能的重投視窗，例如 7 天），或用時間分區表直接 DROP 舊分區

順序性保證

• Kafka：partition 內有序。要讓某實體的事件有序，就以實體 id 為 key。全域順序只能用單一 partition（犧牲平行度）。
• Consumer 端的亂序來源：在 consumer 內用 thread pool 平行處理同 partition 的訊息；重試（失敗的訊息送到 retry topic 後，後面的訊息先被處理）。如需保序，平行處理要以 key 分派到固定 worker（同 key 同 worker）。
• 重試與順序的取捨：嚴格保序時，一則訊息失敗就必須阻塞該 partition（或該 key）直到成功；要不阻塞就得接受亂序，並在 consumer 用版本號處理「舊事件晚到」。
• SQS FIFO / Pub/Sub ordering key / Pulsar Key_Shared：都是「以 key 分組保序」的概念。

Dead Letter Queue 與重試策略

錯誤要分兩類處理：

Kafka 本身沒有「延遲投遞」與「單則重試」，常見做法是 retry topic 階梯（Uber 提出的模式）：

• 主 consumer 不阻塞：失敗的訊息移出主流程，後續訊息繼續處理（代價：該 key 的順序被打破）。
• Retry consumer 讀到訊息時檢查「是否已到可處理時間」，未到就 pause() 該 partition 並等待（因為同一 retry topic 中後面的訊息只會更晚到期），不要 sleep 阻塞 poll 迴圈太久而超過 max.poll.interval.ms。
• 在 header 記錄原始 topic/partition/offset、錯誤原因、重試次數、首次失敗時間，方便追查與 replay。
• DLQ 一定要有監控與告警以及 replay 工具，否則 DLQ 只是資料的墳墓。
• Spring Kafka 的 @RetryableTopic、Kafka Connect 的 errors.deadletterqueue.topic.name 都有內建支援。

import json, time
from confluent_kafka import Producer

RETRY_CHAIN = [("ad-events.retry-1m", 60), ("ad-events.retry-10m", 600)]

class PermanentError(Exception): ...

def route_failure(producer: Producer, msg, err: Exception) -> None:
    headers = dict(msg.headers() or [])
    attempt = int(headers.get("x-retry-attempt", b"0"))
    base = {
        "x-original-topic": headers.get("x-original-topic", msg.topic().encode()),
        "x-original-offset": headers.get("x-original-offset", str(msg.offset()).encode()),
        "x-error": repr(err)[:500].encode(),
    }
    if isinstance(err, PermanentError) or attempt >= len(RETRY_CHAIN):
        target, delay = "ad-events.dlq", 0
    else:
        target, delay = RETRY_CHAIN[attempt]
    base["x-retry-attempt"] = str(attempt + 1).encode()
    base["x-not-before"] = str(int((time.time() + delay) * 1000)).encode()
    producer.produce(target, key=msg.key(), value=msg.value(), headers=base)
    # 注意：要確認轉送成功（flush / delivery callback）後，才 commit 原訊息的 offset

Backpressure

當消費速度長期低於生產速度，佇列只是延後問題。各層的背壓機制：

• Pull 模型天然背壓（Kafka、SQS）：consumer 只拿自己處理得了的量，訊息堆積在 broker（磁碟便宜）而不是 consumer 記憶體。要監控的是 lag。
• Push 模型要靠 prefetch（RabbitMQ）或 flow control（Pub/Sub streaming pull 的 max outstanding messages/bytes）。
• Consumer 內部：用有界佇列連接 poll thread 與 worker pool；佇列滿時 consumer.pause(partitions)，消化後 resume()，期間仍持續呼叫 poll 以維持 group 成員資格。
• Producer 端：buffer 滿時 send() 阻塞或丟 BufferError；collector 收到後應回 503/429 讓 SDK 退避，而不是無限緩衝到 OOM。
• Broker 端：Kafka quota（限制每個 client 的 produce/fetch 速率）、RabbitMQ memory/disk alarm 會阻擋 publisher。

事件驅動架構模式：Event Notification、ECST、Event Sourcing、CQRS、Outbox + CDC 深入

Martin Fowler 把「事件驅動」這個模糊的詞拆成幾種不同的模式，它們的取捨差異很大，面試時能分清楚會很加分。

1. Event Notification（事件通知）

事件只說「發生了什麼」並帶上 id：{"type":"CampaignUpdated","campaign_id":42}。接收方若需要細節，回頭呼叫來源服務的 API 查詢。

• 優點：事件小、來源服務的資料模型不外洩、接收方總是拿到最新資料。
• 缺點：回查造成來源服務負載與耦合（來源掛了接收方也做不了事）；事件與回查之間資料可能又變了。

2. Event-Carried State Transfer（ECST）

事件攜帶完整（或足夠的）狀態：{"type":"CampaignUpdated","campaign_id":42,"budget":1000,"status":"ACTIVE","targeting":{...}}。接收方在本地維護一份副本，不需回查。

• 優點：接收方完全自主（來源服務掛了也能運作）、低延遲讀取。非常適合 bidder：campaign 設定以 ECST 推到各區，出價時只讀本地記憶體。
• 缺點：資料重複存放、最終一致、事件較大；schema 變成服務間契約，演進需要管理。搭配 compacted topic，新的接收方可以重建完整狀態。

3. Event Sourcing

不儲存「目前狀態」，而是儲存所有改變狀態的事件作為唯一真相來源（source of truth），目前狀態由重播事件推導（fold）。例如帳戶不存餘額，而是存所有 Deposited/Withdrawn 事件。

• 優點：完整稽核軌跡、可回到任何時間點的狀態（time travel）、修 bug 後可重新計算、天然產生事件流給其他系統。
• 缺點：學習曲線高；查詢目前狀態需要 projection 或 snapshot；事件 schema 永遠要相容（舊事件不能改）；「刪除個資」（GDPR）困難（常用 crypto shredding：個資加密，刪除金鑰即可）。
• 適合：金流、帳務、需要稽核的領域。不要對整個系統都套用。

4. CQRS（Command Query Responsibility Segregation）

把寫入模型（command，負責驗證與狀態變更）與讀取模型（query，針對查詢最佳化）分開。寫入端發出事件，讀取端訂閱事件建立各種 read model（例如 Elasticsearch 做搜尋、ClickHouse 做報表、Redis 做即時排行榜）。

CQRS 常與 event sourcing 一起出現，但兩者獨立：可以只做 CQRS（寫入用一般 DB，再用 CDC 同步出 read model）。代價是讀取模型最終一致：寫完立刻讀可能讀不到（需要 read-your-writes 的處理，例如寫入後短時間從寫入端讀）。

Outbox Pattern + CDC：解決 Dual Write

服務需要「更新 DB」並「發出事件」。直接兩邊都寫（dual write）的失敗情境：

• DB commit 成功 → 發 Kafka 前 process 被 kill → 事件永遠遺失，下游不一致。
• 先發 Kafka → DB commit 失敗 → 下游收到「幽靈事件」。
• 兩個並行請求 A、B：DB 中 B 後寫，但 Kafka 中 A 後到 → 下游最終狀態與 DB 不同。

兩種 relay 方式：

Debezium 提供 Outbox Event Router（SMT），可以依 outbox 表的欄位把事件路由到對應 topic、以 aggregate_id 當 key。CDC 也可以不用 outbox 直接捕捉業務表變更（例如把 MySQL 的 campaigns 表同步到 Elasticsearch），但這會讓下游依賴你的表結構；outbox 則讓你明確設計對外的事件契約。

CREATE TABLE outbox (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  event_id      CHAR(36)     NOT NULL UNIQUE,     -- 下游去重用
  aggregate_type VARCHAR(64) NOT NULL,            -- 例如 campaign → 路由到 campaign.events topic
  aggregate_id  VARCHAR(64)  NOT NULL,            -- 作為 Kafka key，保證同一實體有序
  event_type    VARCHAR(64)  NOT NULL,
  payload       JSON         NOT NULL,
  created_at    TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3)
);

Schema 演進：Avro、Protobuf 與 Schema Registry 進階

事件一旦發佈，就有多個你不認識的消費者依賴它的格式；而且 Kafka 中保留的舊訊息會被重播。Schema 管理不好的典型災難：producer 把欄位從字串改成數字，所有 consumer 反序列化失敗、lag 爆炸。

格式比較

Schema Registry 與相容性

Confluent Schema Registry（或 Apicurio、AWS Glue Schema Registry）集中保存每個 subject（通常 = topic 名 + -value）的 schema 版本。Producer 序列化時註冊/查找 schema 取得 id，訊息格式為：magic byte (0) + 4 bytes schema id + payload。Consumer 依 id 取回 writer schema 解碼。註冊新版本時，registry 依相容性規則檢查，不相容就拒絕——這是在 CI/部署時就擋下破壞性變更的關鍵。

{
  "type": "record",
  "name": "AdEvent",
  "namespace": "com.example.ads",
  "fields": [
    {"name": "event_id",    "type": "string"},
    {"name": "event_type",  "type": {"type": "enum", "name": "EventType", "symbols": ["IMPRESSION", "CLICK", "CONVERSION"]}},
    {"name": "ts_ms",       "type": {"type": "long", "logicalType": "timestamp-millis"}},
    {"name": "campaign_id", "type": "long"},
    {"name": "user_id",     "type": ["null", "string"], "default": null},
    {"name": "price_micros","type": ["null", "long"],   "default": null, "doc": "v2 新增：有 default，向後相容"}
  ]
}

任務佇列：Celery、RQ 與 MQ 的關係 基礎

任務佇列（task queue）是建立在 MQ 之上的更高層抽象：訊息的內容是「要執行的函式名稱 + 參數」，由 worker 程序取出執行。MQ 關心「傳遞訊息」；task queue 額外提供重試、排程、結果儲存、速率限制、工作流（chain/group/chord）等。

# tasks.py — Celery 可靠性相關設定
from celery import Celery

app = Celery("push", broker="amqp://rabbitmq//", backend="redis://redis/1")
app.conf.update(
    task_acks_late=True,               # 任務「執行完」才 ack（預設是取出即 ack → worker 當機任務遺失）
    task_reject_on_worker_lost=True,   # worker 被 kill 時把任務放回佇列
    worker_prefetch_multiplier=1,      # 每個 worker process 只預取 1 個，長任務時避免囤積
    task_time_limit=300, task_soft_time_limit=270,
    broker_transport_options={"visibility_timeout": 3600},   # 僅 Redis/SQS broker：需大於最長任務時間
)

class TransientError(Exception): ...

@app.task(bind=True, autoretry_for=(TransientError,), retry_backoff=True,
          retry_backoff_max=600, retry_jitter=True, max_retries=5, rate_limit="1000/s")
def send_push(self, campaign_id: int, user_id: str, idem_key: str):
    # acks_late 代表任務可能執行兩次 → 必須冪等
    if already_sent(idem_key):
        return "dup"
    resp = push_provider.send(user_id, build_payload(campaign_id), idempotency_key=idem_key)
    if resp.status in (429, 500, 502, 503):
        raise TransientError(resp.status)
    mark_sent(idem_key)

實戰設計：事件追蹤管線與推播發送佇列 深入 高頻

Appier 真實面試題之一是「客戶端 SDK 回報大量 JSON，設計統計報表系統」。以下聚焦在其中佇列層的設計細節（完整的系統設計流程見 第 13 章，批次處理見 第 12 章）。數字皆為示範用的假設，面試時要先跟面試官確認規模。

SDK 與 Collector 端

• SDK：在裝置端累積事件（例如每 20 則或每 10 秒）批次送出並 gzip，減少請求數與電量；失敗時存在本地、指數退避重送，所以 server 端一定會收到重複 → 每則事件在 client 端產生 event_id（UUID v7）供下游去重。同時帶 client_ts 與 sent_at，server 用 server_ts - sent_at 估算並校正裝置時鐘偏差。
• Collector：無狀態、水平擴展；只做輕量驗證（大小上限、必要欄位、API key）與補充欄位，不要同步寫 DB。Producer 非同步送 Kafka，收到請求就回 202/204。
• Kafka 不可用時：collector 把事件寫進本地磁碟 spool（有上限），恢復後重送；spool 也滿了才回 503 讓 SDK 退避——這就是端到端的 backpressure 鏈。
• Producer 設定：acks=all、idempotence、linger.ms 10–50ms、zstd。曝光事件量大但單筆價值低，可接受稍高延遲換吞吐；轉換事件量小但與計費相關，可靠性優先。

Topic 切分與 Key 選擇

• 依事件類型分 topic：曝光量通常比點擊大一到兩個數量級，轉換更少但最有價值。分開後可以各自設定 partition 數、retention、SLA 與 consumer，避免大量曝光事件拖慢轉換的處理（head-of-line）。
• Key 的選擇取決於下游需要的順序與狀態：
  • 下游要做 per-user 狀態（sessionization、frequency cap、attribution 以 user 為單位 join）→ key = user_id/device_id，同一使用者的事件落在同一 partition，stream processor 的 keyed state 不需 shuffle。
  • 下游只做全域聚合、不需順序 → key = null（sticky partitioner），分佈最均勻。Flink 內部會再依需要 keyBy 重新分區。
  • 避免 key = campaign_id：大型 campaign 的流量可能是小 campaign 的萬倍，造成嚴重 skew 與單一 partition lag。
  • 機器人或異常裝置可能讓單一 user_id 爆量 → 在 collector 做速率限制或把異常流量導到獨立 topic。

Partition 數量估算

經驗公式：partitions ≥ max(目標吞吐 / 單 partition producer 吞吐, 目標吞吐 / 單 partition consumer 吞吐) × 成長與追趕的餘裕。通常 consumer 端的處理才是瓶頸。

1. 假設尖峰曝光 30 萬則/秒、平均 1KB → 約 300MB/s（未壓縮）。
2. 壓測得知單一 consumer 實例處理一個 partition 約 10MB/s（含反序列化、去重、聚合）→ 至少 30 個 partition。
3. 追趕餘裕：若 consumer 停機 1 小時，希望 30 分鐘內追上，處理能力需達輸入速率的 3 倍（C·0.5h = r·(1h + 0.5h)）→ 約 90 個 partition 的平行度。
4. 再考慮未來 1–2 年成長，並選擇有很多因數的數字（便於 consumer 數量整除，負載平均），例如 96 或 120。
5. 不要盲目開到數千：每個 partition 都有檔案 handle、記憶體 buffer、複製流量與 leader 選舉成本；partition 太多會拉長 rebalance 與 failover 時間，並讓 producer batch 變小（每個 partition 各自 batch）。

容量：zstd 壓縮約 5 倍 → 60MB/s → 每天約 5.2TB；RF=3 → 約 15.5TB/天；保留 3 天 → 約 47TB 磁碟，再加上 30–50% 餘裕。長期資料放資料湖（S3/GCS Parquet），Kafka 只保留重播視窗；或使用 tiered storage（Kafka 3.6 起提供，把舊 segment 卸載到物件儲存）。

Lag 告警與 SLO

另外監控：producer 的 error rate 與 record-queue-time、broker 的 under-replicated partitions（URP）與 offline partitions、ISR shrink 頻率、磁碟使用率、請求延遲 p99。

Stream Processor 的關鍵處理

• 去重：以 event_id 為 key 的 state（TTL 數小時到一天）或 Bloom filter 過濾 SDK 重送與 at-least-once 造成的重複。
• Event time 與 watermark：依事件發生時間（而非到達時間）分配視窗；行動裝置離線回補的事件可能晚到數小時，用 watermark + allowed lateness 決定何時輸出結果、晚到資料走 side output 修正。
• Click 與 impression 的 join：以 impression_id 做 interval join（例如點擊在曝光後 30 分鐘內），用於反詐欺與歸因。
• Sink 冪等：聚合結果以 (campaign_id, window_start) upsert，或 ClickHouse ReplacingMergeTree；Flink 以 checkpoint + 交易性 sink 達成端到端 exactly-once。

推播發送佇列設計

情境：行銷自動化平台要對 1,000 萬名使用者發送一則推播活動（類似 AIQUA 的場景），同時也有交易型訊息（OTP、訂單通知）需要秒級送達。Appier 也出過「設計發送 LINE 訊息的 API」的 take-home。

1. Fan-out 分層：一則活動不要變成一則巨大的訊息，也不要由單一程序迴圈 1000 萬次。先由 resolver 分頁查出受眾、切成每批約 1000 人的 job 放進佇列，再由大量 worker 平行處理。批次大小要讓單一 job 在數秒內完成，失敗重試的成本才低。
2. 優先級隔離（bulkhead）：交易型與行銷型分成不同佇列與 worker pool，行銷活動的尖峰不能讓 OTP 延遲。RabbitMQ/SQS 很適合這種「任務型」工作；也可用 Kafka 不同 topic。
3. 速率限制：各 provider 與客戶的 app 都有速率上限，用 Redis 上的分散式 token bucket 控制；收到 429 時遵守 Retry-After 並全域降速（而非每個 worker 各自重試，否則形成 retry storm）。
4. 冪等與去重：以 (campaign_id, user_id) 作為去重 key（Redis SET NX 或 DB 唯一約束），避免 job 重試時重複推播——使用者收到兩則相同推播是很明顯的客訴。呼叫 LINE Messaging API 時可帶 X-Line-Retry-Key（UUID）讓 LINE 端對重試去重。
5. 延遲/排程：依使用者時區在當地 10 點發送、避開靜默時段 → 延遲佇列：Redis ZSET（score = 發送時間，worker 以 ZRANGEBYSCORE 取到期項目）、SQS delay（最長 15 分鐘）、或依小時切分的 topic。
6. 錯誤分類：APNs 回應 token 無效（例如 410 Unregistered）→ 永久錯誤，從裝置表移除 token，不重試；429/5xx → 退避重試；超過重試上限 → DLQ 與告警。
7. 可觀測性：以 campaign 為單位追蹤 queued → sent → delivered → opened 漏斗，佇列深度與發送速率儀表板，預估完成時間；提供暫停/取消活動的開關（worker 每批開始前檢查 campaign 狀態）。

面試自測題