Python 消息队列选型：从 Redis Stream 到 Kafka 的工程决策框架

Python 消息队列选型：从 Redis Stream 到 Kafka 的工程决策框架

一、选型困境：为什么“对的”方案往往也是“错”的

消息队列是分布式系统的核心基础设施，但选型错误导致的架构返工成本极高。常见错误包括：用 Redis List 做消息队列却无法消费确认（消息丢失）、用 RabbitMQ 处理日志流却撑不住吞吐量（单机万级 QPS 上限）、用 Kafka 做任务队列却引入了不必要的运维复杂度。

一个典型的 AI 推理服务场景：API 层接收请求后写入队列，推理 Worker 从队列消费并执行。初期选择 Redis List 实现，上线后发现：消费失败的消息无法重试（List 无 ACK 机制）、消费进度无法追踪（无法知道哪些消息已处理）、队列积压时内存暴涨（Redis 数据全在内存中）。迁移到 Redis Stream 后解决了前两个问题，但第三个问题依然存在。最终迁移到 Kafka 才彻底解决——但迁移成本是重写了整个消费端逻辑。

二、核心特性与选型维度

消息队列的选型不是“哪个更好”，而是“哪个特性组合匹配业务需求”。核心选型维度包括：消息可靠性、吞吐量、消费模型、运维复杂度和生态集成。

flowchart TB
A[消息队列选型维度] –> B[消息可靠性]
A –> C[吞吐量]
A –> D[消费模型]
A –> E[运维复杂度]

B –> B1[至少一次: Redis Stream]
B –> B2[精确一次: Kafka]
B –> B3[最多一次: Redis Pub/Sub]

C –> C1[万级 QPS: RabbitMQ]
C –> C2[十万级 QPS: Redis Stream]
C –> C3[百万级 QPS: Kafka]

D –> D1[点对点: Queue 模型]
D –> D2[发布订阅: Topic 模型]
D –> D3[消费者组: Group 模型]

E –> E1[轻量: Redis / RabbitMQ]
E –> E2[重量: Kafka / Pulsar]

B1 –> F{业务场景匹配}
C2 –> F
D1 –> F

F –>|任务队列, 低吞吐| G[Redis Stream]
F –>|事件流, 高吞吐| H[Kafka]
F –>|复杂路由, 低延迟| I[RabbitMQ]

2.1 消息可靠性等级

• 最多一次（At Most Once）：消息可能丢失，但不会重复。适用于日志、指标等可丢失场景。Redis Pub/Sub 属于此类。
• 至少一次（At Least Once）：消息不会丢失，但可能重复。适用于任务队列，消费端需要幂等处理。Redis Stream、RabbitMQ 属于此类。
• 精确一次（Exactly Once）：消息既不丢失也不重复。Kafka 通过事务和幂等生产者实现，但代价是吞吐量下降约 20%。

2.2 消费模型

• 点对点（Queue）：一条消息只能被一个消费者处理。适用于任务分发场景。
• 发布订阅（Topic）：一条消息被所有订阅者接收。适用于事件通知场景。
• 消费者组（Consumer Group）：同一组内的消费者竞争消费，不同组独立消费。Kafka 和 Redis Stream 均支持。

2.3 Python 客户端生态

• Redis：redis-py（同步）/ aioredis（异步），成熟稳定
• RabbitMQ：pika（同步）/ aio-pika（异步），功能完整
• Kafka：confluent-kafka-python（基于 librdkafka，高性能）/ kafka-python（纯 Python，性能差）

三、Python 消息队列的代码实现

3.1 Redis Stream 消费者

import redis
import json
import time
from typing import Callable, Optional
from dataclasses import dataclass

@dataclass
class StreamMessage:
msg_id: str
data: dict
stream: str
consumer_group: str

class RedisStreamConsumer:
def __init__(self, redis_url: str = "redis://localhost:6379",
consumer_group: str = "default",
consumer_name: str = "worker-1"):
self.client = redis.from_url(redis_url)
self.consumer_group = consumer_group
self.consumer_name = consumer_name

def ensure_group(self, stream: str) -> None:
try:
self.client.xgroup_create(
stream, self.consumer_group, id="0", mkstream=True
)
except redis.ResponseError as e:
if "BUSYGROUP" not in str(e):
raise

def consume_loop(self, stream: str,
handler: Callable[[StreamMessage], bool],
batch_size: int = 10,
block_ms: int = 5000) -> None:
self.ensure_group(stream)

while True:
messages = self.client.xreadgroup(
self.consumer_group,
self.consumer_name,
{stream: ">"},
count=batch_size,
block=block_ms,
)

if not messages:
self._claim_pending(stream, handler)
continue

for stream_name, msg_list in messages:
for msg_id, data in msg_list:
msg = StreamMessage(
msg_id=msg_id.decode(),
data={k.decode(): v.decode()
for k, v in data.items()},
stream=stream_name.decode(),
consumer_group=self.consumer_group,
)

try:
success = handler(msg)
if success:
self.client.xack(
stream, self.consumer_group, msg_id
)
except Exception as e:
print(f"消息处理失败: {msg_id}, 错误: {e}")

def _claim_pending(self, stream: str,
handler: Callable[[StreamMessage], bool],
min_idle_ms: int = 60000) -> None:
pending = self.client.xpending_range(
stream, self.consumer_group,
min="-", max="+", count=10,
)

if not pending:
return

pending_ids = [p["message_id"] for p in pending
if p.get("time_since_delivered", 0) > min_idle_ms]

if not pending_ids:
return

claimed = self.client.xclaim(
stream, self.consumer_group,
self.consumer_name, min_idle_ms,
pending_ids,
)

for msg_id, data in claimed:
msg = StreamMessage(
msg_id=msg_id.decode(),
data={k.decode(): v.decode()
for k, v in data.items()},
stream=stream,
consumer_group=self.consumer_group,
)
try:
success = handler(msg)
if success:
self.client.xack(stream, self.consumer_group, msg_id)
except Exception as e:
print(f"重试消息处理失败: {msg_id}, 错误: {e}")

3.2 Kafka 生产者与消费者

from confluent_kafka import Producer, Consumer, KafkaError
import json
from typing import Callable

class KafkaProducer:
def __init__(self, bootstrap_servers: str = "localhost:9092"):
self.producer = Producer({
"bootstrap.servers": bootstrap_servers,
"enable.idempotence": True,
"acks": "all",
"retries": 3,
"batch.size": 16384,
"linger.ms": 5,
})

def send(self, topic: str, key: str, value: dict) -> None:
self.producer.produce(
topic=topic,
key=key.encode("utf-8"),
value=json.dumps(value, ensure_ascii=False).encode("utf-8"),
callback=self._delivery_report,
)
self.producer.poll(0)

def flush(self, timeout: float = 10.0) -> None:
remaining = self.producer.flush(timeout)
if remaining > 0:
print(f"警告: {remaining} 条消息未发送完成")

@staticmethod
def _delivery_report(err, msg):
if err:
print(f"消息发送失败: {err}, Topic: {msg.topic()}")

class KafkaConsumerWrapper:
def __init__(self, bootstrap_servers: str = "localhost:9092",
group_id: str = "default"):
self.consumer = Consumer({
"bootstrap.servers": bootstrap_servers,
"group.id": group_id,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"max.poll.records": 100,
})

def consume_loop(self, topics: list[str],
handler: Callable[[dict], bool],
poll_timeout: float = 1.0) -> None:
self.consumer.subscribe(topics)

while True:
msg = self.consumer.poll(poll_timeout)

if msg is None:
continue

if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
print(f"Kafka 消费错误: {msg.error()}")
continue

try:
value = json.loads(msg.value().decode("utf-8"))
success = handler(value)

if success:
self.consumer.commit(asynchronous=False)
except Exception as e:
print(f"消息处理失败: {e}, Offset: {msg.offset()}")

def close(self):
self.consumer.close()

3.3 选型决策工具

from dataclasses import dataclass

@dataclass
class QueueRequirement:
throughput_qps: int
message_size_kb: int
reliability: str
consumer_model: str
message_retention: str
ordering: bool
ops_complexity: str

def recommend_queue(req: QueueRequirement) -> list[dict]:
recommendations = []

if req.throughput_qps <= 100_000 and req.reliability in (
"at-most-once", "at-least-once"
):
score = 80
if req.ops_complexity == "low":
score += 10
if req.consumer_model in ("queue", "consumer-group"):
score += 5
recommendations.append({
"queue": "Redis Stream",
"score": score,
"reason": "轻量部署，支持消费者组和 ACK，适合任务队列",
"caveat": "数据全在内存，积压时内存压力大",
})

if req.throughput_qps <= 50_000 and req.consumer_model in (
"queue", "pub-sub"
):
score = 70
if req.ops_complexity == "low":
score += 5
recommendations.append({
"queue": "RabbitMQ",
"score": score,
"reason": "丰富的路由和交换机类型，支持消息确认和死信队列",
"caveat": "单机吞吐量上限约 5 万 QPS，集群扩展复杂",
})

if req.throughput_qps > 10_000 or req.message_retention in (
"days", "weeks"
):
score = 85
if req.reliability == "exactly-once":
score += 5
if req.ordering:
score += 5
recommendations.append({
"queue": "Kafka",
"score": score,
"reason": "高吞吐、持久化存储、支持精确一次和消息顺序",
"caveat": "运维复杂度高，需要 ZooKeeper/KRaft 集群",
})

recommendations.sort(key=lambda x: x["score"], reverse=True)
return recommendations

四、架构权衡

权衡一：内存与磁盘。Redis 数据在内存中，积压时内存暴涨；Kafka 数据在磁盘上，积压时磁盘空间增长但内存稳定。对于消息量波动大的场景，Kafka 更安全。

权衡二：简单性与功能性。Redis Stream 最简单（复用已有 Redis），但功能有限（无死信队列、无延迟消息）；RabbitMQ 功能最丰富（延迟队列、死信、路由），但性能最低；Kafka 性能最强，但运维最复杂。

权衡三：消费幂等性。至少一次投递意味着消息可能重复，消费端必须实现幂等处理。常见方案：用消息 ID 去重（Redis SET 记录已处理 ID）、用数据库唯一约束防止重复写入。

五、总结

消息队列选型的核心思路是“需求驱动选型，而非技术驱动选型”。低吞吐任务队列选 Redis Stream，复杂路由场景选 RabbitMQ，高吞吐事件流选 Kafka——每种队列都有其最优的适用场景。

落地步骤：第一步，明确业务需求（吞吐量、可靠性、消费模型、消息保留时间）；第二步，用选型决策工具生成推荐列表，选择得分最高的方案；第三步，在测试环境验证选型方案的吞吐量和可靠性，确认满足预期。关键原则是——不要为了“技术先进”而选择 Kafka，也不要为了“简单”而忽视可靠性需求，选型的唯一标准是业务需求。