消息队列选型：Kafka、RabbitMQ 与 Redis 对比分析

三维度对比

实战代码示例

Kafka 生产者与消费者

from kafka import KafkaProducer, KafkaConsumer
import json

class KafkaMQ:
    def __init__(self, servers):
        self.producer = KafkaProducer(
            bootstrap_servers=servers,
            acks='all',
            value_serializer=lambda v: json.dumps(v).encode()
        )
    
    def send(self, topic, data):
        future = self.producer.send(topic, value=data)
        print(f"已发送：{future.get(timeout=10)}")

注：生产环境建议配置重试机制与批量发送参数

RabbitMQ 消息路由

import pika

def publish_message(conn, exchange, routing_key, message):
    channel = conn.channel()
    channel.basic_publish(
        exchange=exchange,
        routing_key=routing_key,
        body=json.dumps(message),
        properties=pika.BasicProperties(delivery_mode=2)
    )

提示：delivery_mode=2 确保消息持久化

Redis Stream 消费组

def consume_stream(client, stream, group, consumer):
    messages = client.xreadgroup(
        group, consumer,
        {stream: '>'},
        count=10, block=5000
    )
    for _, msgs in messages:
        for mid, fields in msgs:
            process(fields)
            client.xack(stream, group, mid)

注意：及时 XACK 避免消息滞留 PEL

选型决策指南

日志收集 → Kafka（高吞吐 + 回溯） 订单处理 → RabbitMQ（可靠路由 + 延迟队列） 实时通知 → Redis Pub/Sub（毫秒延迟） 混合架构：电商场景中 Kafka 处理日志，RabbitMQ 管理订单，Redis 支撑 WebSocket 通信

总结

消息队列选型需权衡业务需求与技术特性：

1. 大数据场景优先 Kafka 的吞吐与回溯能力
2. 复杂业务流选择 RabbitMQ 的路由灵活性
3. 轻量实时需求用 Redis 降低运维成本
4. 云原生趋势可考虑托管服务（如 AWS SQS）

实际部署中建议结合监控工具（如 Prometheus）持续优化配置参数，根据业务增长动态调整架构。