投递保证：消息不丢失、不重复
在 coomia-dip 的多 Layer 架构中，Layer 之间通过异步消息通信。但异步消息面临多种故障：
Coomia发布于 2025年12月25日10 分钟阅读
分享本文Twitter / X
投递保证：消息不丢失、不重复
“系列：S10 设计模式 · 第 10 篇 | 难度：高级 | 阅读时间：18 分钟
#TL;DR
- 投递保证（Delivery Guarantee）模式确保跨 Layer 的异步消息在各种故障场景下不丢失、不重复处理。coomia-dip 在不同场景下使用三种保证级别：至少一次（At-Least-Once）、至多一次（At-Most-Once）和恰好一次（Exactly-Once）。
- coomia-dip 通过 Kafka 事务性生产者、幂等消费者、Outbox 模式和死信队列（DLQ）实现端到端的消息可靠性。
- 结合 Transactional Outbox 和 CDC（Change Data Capture），coomia-dip 实现了数据库写入与消息发送的原子性保证。
#引言：分布式系统中的消息可靠性
在 coomia-dip 的多 Layer 架构中，Layer 之间通过异步消息通信。但异步消息面临多种故障：
Code
故障 1：生产者崩溃——消息已写入业务数据库，但消息还没发送到 Kafka
故障 2：Kafka 节点故障——消息已发送但未被确认
故障 3：消费者崩溃——消息已从 Kafka 读取但未完成处理
故障 4：网络分区——生产者和 Kafka 之间的网络中断
故障 5：重复投递——Kafka 重试导致消息被发送两次
不同的业务场景对消息可靠性有不同的要求。投递保证模式为每种场景选择合适的保证级别。
#一、三种保证级别
#1.1 At-Most-Once（至多一次）
消息最多被处理一次。可能丢失，但绝不重复。适用于丢失可容忍的场景（如监控指标采集）。
Python
class AtMostOnceProducer:
"""Producer with at-most-once delivery guarantee."""
async def send(self, topic: str, message: dict) -> None:
"""Send message with no retry — fire and forget."""
try:
await self._kafka_producer.send(
topic=topic,
value=self._serialize(message),
acks=0, # 不等待确认
)
except Exception:
pass # 发送失败就丢弃
class AtMostOnceConsumer:
"""Consumer with at-most-once delivery guarantee."""
async def consume(self, topic: str, handler: callable) -> None:
"""Commit offset before processing — may lose on crash."""
async for message in self._kafka_consumer.subscribe(topic):
# 先提交 offset，再处理
await self._kafka_consumer.commit(message.offset)
try:
await handler(message.value)
except Exception:
# 处理失败，消息已经提交，不会重试
pass
#1.2 At-Least-Once（至少一次）
消息至少被处理一次。不会丢失，但可能重复。适用于大多数业务场景，配合幂等消费者可以达到 Exactly-Once 效果。
Python
class AtLeastOnceProducer:
"""Producer with at-least-once delivery guarantee."""
async def send(self, topic: str, message: dict) -> None:
"""Send message with retries until acknowledged."""
max_retries = 5
for attempt in range(max_retries):
try:
await self._kafka_producer.send(
topic=topic,
value=self._serialize(message),
acks="all", # 等待所有副本确认
)
return
except Exception as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
class AtLeastOnceConsumer:
"""Consumer with at-least-once delivery guarantee."""
async def consume(self, topic: str, handler: callable) -> None:
"""Process first, then commit — may reprocess on crash."""
async for message in self._kafka_consumer.subscribe(topic):
try:
# 先处理，再提交
await handler(message.value)
await self._kafka_consumer.commit(message.offset)
except Exception:
# 处理失败，不提交，下次重新消费
await self._error_handler.handle(message)
#1.3 Exactly-Once（恰好一次）
消息恰好被处理一次。不丢失，不重复。coomia-dip 通过 Kafka 事务 + 幂等消费者实现：
Python
class ExactlyOnceProducer:
"""Producer with exactly-once delivery guarantee."""
def __init__(self, transactional_id: str):
self._producer = KafkaProducer(
transactional_id=transactional_id,
enable_idempotence=True,
acks="all",
)
async def send_transactional(
self,
messages: list[tuple[str, dict]],
) -> None:
"""Send multiple messages in a single transaction."""
await self._producer.begin_transaction()
try:
for topic, message in messages:
await self._producer.send(topic, self._serialize(message))
await self._producer.commit_transaction()
except Exception:
await self._producer.abort_transaction()
raise
class ExactlyOnceConsumer:
"""Consumer with exactly-once via idempotent processing."""
def __init__(self, idempotency_store: "IdempotencyStore"):
self._idempotency = idempotency_store
async def consume(self, topic: str, handler: callable) -> None:
"""Process with idempotency check to ensure exactly-once."""
async for message in self._kafka_consumer.subscribe(topic):
message_id = message.headers.get("message_id")
# 幂等性检查
if await self._idempotency.is_processed(message_id):
# 已经处理过，跳过并提交
await self._kafka_consumer.commit(message.offset)
continue
try:
await handler(message.value)
await self._idempotency.mark_processed(message_id)
await self._kafka_consumer.commit(message.offset)
except Exception:
await self._error_handler.handle(message)
#二、Transactional Outbox 模式
#2.1 问题：双写一致性
业务操作通常需要同时写数据库和发消息。两个操作不在同一个事务中，可能导致不一致：
Code
场景：创建订单后发送通知
1. 写入订单到数据库 ✓
2. 发送消息到 Kafka ✗ (网络故障)
结果：订单已创建但通知未发送
#2.2 Outbox 解决方案
将消息写入数据库的 outbox 表（与业务数据在同一个事务中），然后由独立的进程从 outbox 表读取并发送到 Kafka：
Python
class TransactionalOutbox:
"""Ensure atomic database write and message send."""
async def execute_with_outbox(
self,
db_operation: callable,
messages: list[dict],
) -> None:
"""Execute DB operation and enqueue messages atomically."""
async with self._db.transaction() as tx:
# 执行业务数据库操作
result = await db_operation(tx)
# 将消息写入 outbox 表（同一事务）
for msg in messages:
await tx.execute(
"INSERT INTO outbox (message_id, topic, payload, status, created_at) "
"VALUES ($1, $2, $3, 'pending', NOW())",
generate_id(), msg["topic"], json.dumps(msg["payload"]),
)
# 事务提交时，业务数据和 outbox 消息一起持久化
return result
class OutboxRelay:
"""Relay messages from outbox table to Kafka."""
async def relay_pending(self) -> int:
"""Poll outbox table and send pending messages to Kafka."""
pending = await self._db.fetch(
"SELECT * FROM outbox WHERE status = 'pending' "
"ORDER BY created_at LIMIT 100"
)
sent_count = 0
for msg in pending:
try:
await self._kafka_producer.send(
topic=msg["topic"],
value=msg["payload"],
headers={"message_id": msg["message_id"]},
)
await self._db.execute(
"UPDATE outbox SET status = 'sent', sent_at = NOW() "
"WHERE message_id = $1",
msg["message_id"],
)
sent_count += 1
except Exception as e:
await self._db.execute(
"UPDATE outbox SET status = 'failed', error = $2, "
"retry_count = retry_count + 1 WHERE message_id = $1",
msg["message_id"], str(e),
)
return sent_count
async def run_relay_loop(self) -> None:
"""Continuously relay outbox messages."""
while True:
sent = await self.relay_pending()
if sent == 0:
await asyncio.sleep(1) # 没有待发送消息时等待
#2.3 基于 CDC 的 Outbox
更高效的方案是使用 CDC（Change Data Capture）监听 outbox 表的变更，而不是轮询：
Python
class CDCOutboxRelay:
"""Use CDC to capture outbox changes instead of polling."""
async def start(self) -> None:
"""Start CDC listener for outbox table changes."""
async for change in self._cdc_client.subscribe("outbox"):
if change.operation == "INSERT":
await self._kafka_producer.send(
topic=change.new_row["topic"],
value=change.new_row["payload"],
headers={"message_id": change.new_row["message_id"]},
)
await self._db.execute(
"UPDATE outbox SET status = 'sent' WHERE message_id = $1",
change.new_row["message_id"],
)
#三、幂等消费者
#3.1 幂等性存储
Python
class IdempotencyStore:
"""Store for tracking processed message IDs."""
async def is_processed(self, message_id: str) -> bool:
"""Check if a message has already been processed."""
result = await self._redis.get(f"idempotency:{message_id}")
return result is not None
async def mark_processed(
self, message_id: str, ttl_hours: int = 168
) -> None:
"""Mark a message as processed with TTL."""
await self._redis.set(
f"idempotency:{message_id}",
"1",
ex=ttl_hours * 3600,
)
async def mark_processed_with_result(
self, message_id: str, result: dict, ttl_hours: int = 168
) -> None:
"""Mark as processed and store the result for replay."""
await self._redis.set(
f"idempotency:{message_id}",
json.dumps(result),
ex=ttl_hours * 3600,
)
async def get_result(self, message_id: str) -> dict | None:
"""Get the stored result of a previously processed message."""
result = await self._redis.get(f"idempotency:{message_id}")
if result and result != "1":
return json.loads(result)
return None
#3.2 业务级幂等
某些业务操作天然是幂等的（如设置状态），而某些不是（如扣减余额）。coomia-dip 为非幂等操作提供幂等包装：
Python
class BusinessIdempotency:
"""Make non-idempotent business operations idempotent."""
async def debit_account_idempotent(
self,
idempotency_key: str,
account_id: str,
amount: float,
) -> dict:
"""Idempotent account debit using idempotency key."""
# 检查是否已执行
existing = await self._idempotency.get_result(idempotency_key)
if existing:
return existing # 返回之前的结果
# 执行扣减
result = await self._account_service.debit(account_id, amount)
# 记录结果
await self._idempotency.mark_processed_with_result(
idempotency_key, result
)
return result
#四、死信队列（DLQ）
#4.1 DLQ 策略
消费者反复处理失败的消息进入死信队列，避免阻塞正常消息处理：
Python
class DeadLetterQueueHandler:
"""Handle messages that cannot be processed."""
def __init__(self, max_retries: int = 3, retry_delays: list[int] | None = None):
self._max_retries = max_retries
self._retry_delays = retry_delays or [60, 300, 3600] # 1min, 5min, 1hr
async def handle_failure(
self,
message: dict,
error: Exception,
retry_count: int,
) -> None:
"""Handle a failed message processing attempt."""
if retry_count < self._max_retries:
# 发送到重试队列（延迟投递）
delay = self._retry_delays[min(retry_count, len(self._retry_delays) - 1)]
await self._kafka_producer.send(
topic=f"{message['topic']}.retry",
value=message,
headers={
"retry_count": str(retry_count + 1),
"original_topic": message["topic"],
"retry_at": str(time.time() + delay),
"error": str(error),
},
)
else:
# 超过最大重试次数，进入死信队列
await self._kafka_producer.send(
topic=f"{message['topic']}.dlq",
value=message,
headers={
"retry_count": str(retry_count),
"original_topic": message["topic"],
"final_error": str(error),
"failed_at": datetime.utcnow().isoformat(),
},
)
# 通知运维
await self._alert_service.send(
severity="warning",
title=f"Message sent to DLQ: {message['topic']}",
message=f"Message {message.get('message_id')} failed after {retry_count} retries: {error}",
)
async def replay_dlq(
self, topic: str, filter_fn: callable | None = None
) -> int:
"""Replay messages from dead letter queue."""
replayed = 0
dlq_topic = f"{topic}.dlq"
async for message in self._kafka_consumer.subscribe(dlq_topic):
if filter_fn and not filter_fn(message):
continue
# 重新发送到原始队列
await self._kafka_producer.send(
topic=message.headers["original_topic"],
value=message.value,
headers={"replayed_from_dlq": "true"},
)
replayed += 1
return replayed
#五、coomia-dip 中的投递保证配置
#5.1 按场景选择保证级别
Python
DELIVERY_CONFIG = {
# 核心业务操作：Exactly-Once
"ontology.action.execute": {
"guarantee": "exactly_once",
"use_outbox": True,
"idempotency_ttl_hours": 168,
"dlq_max_retries": 5,
},
# 状态变更通知：At-Least-Once
"ontology.state.changed": {
"guarantee": "at_least_once",
"use_outbox": True,
"idempotency_ttl_hours": 24,
"dlq_max_retries": 3,
},
# 审计日志：At-Least-Once
"audit.event": {
"guarantee": "at_least_once",
"use_outbox": False,
"dlq_max_retries": 10,
},
# 监控指标：At-Most-Once
"metrics.collect": {
"guarantee": "at_most_once",
"use_outbox": False,
},
# 推理结果缓存刷新：At-Most-Once
"reasoning.cache.invalidate": {
"guarantee": "at_most_once",
"use_outbox": False,
},
}
#5.2 端到端消息追踪
Python
class MessageTracer:
"""End-to-end message tracing for debugging delivery issues."""
async def trace_message(self, message_id: str) -> dict:
"""Trace a message through the entire delivery pipeline."""
return {
"message_id": message_id,
"produced_at": await self._get_produce_time(message_id),
"kafka_partition": await self._get_partition(message_id),
"consumed_at": await self._get_consume_time(message_id),
"processed": await self._idempotency.is_processed(message_id),
"retry_count": await self._get_retry_count(message_id),
"in_dlq": await self._is_in_dlq(message_id),
"outbox_status": await self._get_outbox_status(message_id),
}
#六、消息顺序性保证
#6.1 分区级顺序
Python
class OrderedMessageProducer:
"""Produce messages with ordering guarantee within a partition."""
async def send_ordered(
self,
topic: str,
ordering_key: str,
message: dict,
) -> None:
"""Send message to specific partition based on ordering key."""
partition = self._hash_to_partition(ordering_key, self._num_partitions)
await self._kafka_producer.send(
topic=topic,
value=self._serialize(message),
partition=partition,
key=ordering_key.encode(),
)
def _hash_to_partition(self, key: str, num_partitions: int) -> int:
"""Consistently hash a key to a partition number."""
import hashlib
hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
return hash_val % num_partitions
#Key Takeaways
- 三级保证：At-Most-Once、At-Least-Once、Exactly-Once 适用于不同业务场景
- Transactional Outbox：通过 Outbox 表解决数据库写入和消息发送的双写一致性问题
- 幂等消费者：幂等性检查是实现 Exactly-Once 语义的关键
- 死信队列：DLQ + 重试策略确保问题消息不阻塞正常处理流程
- CDC 优化：基于 CDC 的 Outbox Relay 比轮询更高效、延迟更低
- 分区顺序：通过分区键保证相关消息的处理顺序
#Next Article
下一篇我们将探讨门面模式（Facade Pattern）——coomia-dip 如何为复杂的多 Layer 架构提供简洁统一的 API 入口。
S10-11: 门面模式：统一 API 入口
#Tags
#设计模式 #投递保证 #DeliveryGuarantee #ExactlyOnce #Outbox #CDC #幂等 #死信队列 #Kafka