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

Kafka 的分区机制是其高吞吐量的关键。每个 Topic 可以划分为多个 Partition，分布在不同 Broker 上。生产者可以并行向多个 Partition 写入数据，消费者组内的消费者也可以并行从不同 Partition 消费数据。这种设计充分利用了多核 CPU 和多节点的计算能力。

2.3 Kafka 消息模型

Kafka 采用 发布 - 订阅模型，同时支持 拉取模式 消费。

发布 - 订阅模型：

生产者将消息发送到特定 Topic，不关心谁会消费这些消息。消费者订阅感兴趣的 Topic，可以属于不同的消费者组。同一消费者组内的消费者共同分担消息处理（每条消息只被组内一个消费者处理），不同消费者组独立消费（每条消息会被每个组各消费一次）。

拉取模式：

Kafka 消费者主动从 Broker 拉取消息，而非 Broker 推送。这种设计有几个优势：

1. 流量控制：消费者可以根据自身处理能力控制拉取速率
2. 批量处理：消费者可以批量拉取消息，提高效率
3. 断点续传：消费者记录 Offset，故障恢复后可以继续消费

2.4 Kafka 一致性保证

Kafka 提供了多层次的一致性保证机制：

消息持久化：

Kafka 将消息持久化到磁盘，而非内存。消息以顺序写的方式追加到日志文件，这种设计充分利用了磁盘的顺序写性能（可达数百 MB/s），远超随机写性能。消息保留策略可配置，支持基于时间和大小的保留策略。

ACK 机制：

生产者发送消息时可以配置 ACK 级别：

副本机制：

每个 Partition 可以配置多个副本，分布在不同 Broker 上。副本分为 Leader 和 Follower，Leader 负责读写，Follower 被动同步。当 Leader 故障时，从 ISR（In-Sync Replicas）中选举新的 Leader，确保数据不丢失。

Exactly-Once 语义：

Kafka 支持三种消息语义：

• At Most Once：消息可能丢失，但不会重复
• At Least Once：消息不会丢失，但可能重复
• Exactly Once：消息不丢失不重复（需要配合幂等性生产者和事务机制）

3. RabbitMQ 详解 - 架构、消息模型、一致性保证

3.1 RabbitMQ 的发展历程

RabbitMQ 是一款开源的消息代理软件，由 Pivotal Software（现属 VMware）维护。它实现了 AMQP（Advanced Message Queuing Protocol）协议，最早于 2007 年发布。RabbitMQ 的设计灵感来源于电信系统的消息交换机制，其名称中的 Rabbit 象征着消息的快速传递和繁殖。

RabbitMQ 以其可靠性、灵活性和易用性著称，在企业级应用中得到广泛应用。它支持多种消息协议，提供丰富的客户端 SDK，是传统企业应用集成的首选方案。

3.2 RabbitMQ 架构设计

RabbitMQ 采用经典的消息代理架构，核心组件包括 Producer、Exchange、Queue、Consumer。

核心概念解析：

RabbitMQ 的 Exchange 是其灵活性的核心。不同类型的 Exchange 提供不同的路由策略，满足各种业务场景需求。

3.3 RabbitMQ 消息模型

RabbitMQ 支持多种消息模型，通过不同类型的 Exchange 实现：

Exchange 类型对比：

推模式消费：

RabbitMQ 默认采用推模式，Broker 主动将消息推送给消费者。消费者注册回调函数，当消息到达时自动触发。这种模式响应及时，但需要消费者有足够的处理能力，否则可能导致消息堆积。

3.4 RabbitMQ 一致性保证

RabbitMQ 提供了完善的消息可靠性机制：

消息持久化：

消息持久化需要三个条件同时满足：

1. Exchange 持久化：声明 Exchange 时设置 durable=true
2. Queue 持久化：声明 Queue 时设置 durable=true
3. Message 持久化：发送消息时设置 delivery_mode=2

ACK 机制：

RabbitMQ 支持自动确认和手动确认两种模式：

• 自动确认（Auto ACK）：消息发送给消费者后立即删除，性能高但可能丢失消息
• 手动确认（Manual ACK）：消费者处理完成后发送 ACK，消息才删除，可靠性高

消费者还可以发送 NACK（否定确认），要求 Broker 重新入队或丢弃消息。

镜像队列：

RabbitMQ 支持镜像队列（Mirrored Queue），将队列复制到多个节点。当主节点故障时，从镜像节点选举新的主节点，实现高可用。但镜像队列会影响性能，且不支持横向扩展。

消息确认模式：

4. Redis 消息队列 - 架构、消息模型、一致性保证

4.1 Redis 作为消息队列的定位

Redis 本质上是一个高性能的内存数据库，但其丰富的数据结构使其也能胜任消息队列的角色。Redis 作为消息队列并非不务正业，而是在特定场景下的最优选择——当你已经使用 Redis 作为缓存，且消息队列需求简单时，复用 Redis 可以减少系统复杂度。

Redis 消息队列的优势在于：

• 极低延迟：内存操作，延迟在毫秒级
• 部署简单：无需额外组件，复用现有 Redis
• 功能够用：支持发布订阅、延迟队列、消息持久化

4.2 Redis 消息队列架构

Redis 提供了多种消息队列实现方式，从简单到复杂依次为：

4.3 Redis Stream 详解

Redis 5.0 引入的 Stream 数据结构是 Redis 对消息队列的正式支持。Stream 借鉴了 Kafka 的设计理念，提供了完善的消息队列功能：

核心概念：

消息 ID 设计：

Stream 的消息 ID 格式为 <millisecondsTime>-<sequenceNumber>，例如 1638240000000-0。这种设计保证了消息的全局有序性，且 ID 本身携带时间信息。

4.4 Redis 一致性保证

Redis Stream 提供了基本的一致性保证：

消息持久化：

Redis 支持 RDB 和 AOF 两种持久化方式：

• RDB：定时快照，可能丢失最近的数据
• AOF：追加日志，数据更安全但性能略低

ACK 机制：

Stream 通过 XACK 命令确认消息。消费者读取消息后，消息进入 PEL（Pending Entries List），只有发送 XACK 后才从 PEL 中移除。如果消费者故障，可以通过 XPENDING 查看未确认消息，重新分配给其他消费者。

消费者组：

Stream 支持消费者组，组内消费者共同消费消息。每个消息只会被组内一个消费者处理，实现负载均衡。同时支持消息重平衡和故障转移。

局限性：

Redis 作为消息队列的局限性需要清醒认识：

• 内存限制：消息存储在内存，容量受限于内存大小
• 持久化风险：AOF/RDB 可能丢失数据
• 功能有限：相比 Kafka/RabbitMQ，缺少死信队列、延迟队列等高级功能

5. 三者对比分析 - 性能、可靠性、场景适用性

5.1 架构对比

三种消息队列在设计理念上存在根本差异：

5.2 性能对比

性能是选型的关键考量因素，以下是典型场景下的性能对比：

性能分析：

• Kafka：磁盘顺序写优化，大消息吞吐量优势明显，适合大数据场景
• RabbitMQ：内存操作为主，小消息延迟低，适合实时性要求高的场景
• Redis：纯内存操作，延迟最低，但受内存容量限制

5.3 可靠性对比

可靠性是消息队列的生命线，三者在这方面各有侧重：

5.4 功能特性对比

5.5 运维复杂度对比

6. 场景选型指南 - 不同业务场景的最佳选择

6.1 日志收集场景

场景特点：

• 数据量大，吞吐量要求高
• 消息丢失可容忍（日志可重新生成）
• 需要消息回溯和重放
• 消费者数量多，需要并行处理

推荐方案：Kafka

Kafka 的设计初衷就是日志收集，天然适合这个场景：

• 高吞吐量：单节点可达百万级消息/秒
• 消息持久化：消息持久化到磁盘，支持长期存储
• 消息回溯：消费者可以重置 Offset，重新消费历史消息
• 水平扩展：通过增加 Partition 和 Broker 轻松扩展

6.2 订单处理场景

场景特点：

• 消息不能丢失，可靠性要求高
• 业务逻辑复杂，需要灵活路由
• 需要延迟队列（订单超时取消）
• 消息量中等，实时性要求高

推荐方案：RabbitMQ

RabbitMQ 的可靠性保证和丰富功能完美匹配订单场景：

• 消息可靠性：完善的 ACK 机制，确保消息不丢失
• 灵活路由：Topic Exchange 支持多条件路由
• 延迟队列：通过插件实现订单超时取消
• 死信队列：处理失败订单，支持人工干预

典型流程：

1. 订单创建 → 发送到订单队列
2. 支付成功 → 发送到发货队列
3. 支付超时 → 延迟队列触发取消
4. 处理失败 → 进入死信队列

6.3 实时通信场景

场景特点：

• 消息实时性要求极高
• 消息量小但频繁
• 在线状态感知
• 已有 Redis 缓存基础设施

推荐方案：Redis Pub/Sub

Redis 的极低延迟使其成为实时通信的理想选择：

• 毫秒级延迟：内存操作，响应极快
• 简单高效：Pub/Sub 模式简单易用
• 复用设施：无需额外部署消息队列
• 在线感知：结合 Redis 存储用户在线状态

注意事项：

• Pub/Sub 不持久化，离线用户收不到消息
• 需要配合其他存储记录历史消息
• 大量订阅时注意内存使用

6.4 选型决策树

• 消息量级？百万级/天以上 -> Kafka
• 是否需要消息回溯？是 -> Kafka / Redis Stream
• 可靠性要求？高 -> RabbitMQ / Kafka
• 实时性要求？极高 -> Redis Stream
• 需要延迟队列？是 -> RabbitMQ / Redis Stream
• 已有 Redis？是 -> Redis Stream

6.5 混合架构实践

在实际项目中，单一消息队列往往无法满足所有需求。混合架构是更常见的选择：

典型案例：电商系统

┌─────────────────────────────────────────────────────────┐
│ 电商系统架构                                            │
├─────────────────────────────────────────────────────────┤
│ 日志收集层                                              │
│ Kafka 集群（高吞吐量）                                  │
├─────────────────────────────────────────────────────────┤
│ 业务处理层                                              │
│ RabbitMQ（订单、支付、库存）                            │
├─────────────────────────────────────────────────────────┤
│ 实时通信层                                              │
│ Redis Pub/Sub（WebSocket、通知）                        │
└─────────────────────────────────────────────────────────┘

7. 实战代码示例 - Python 连接三种消息队列

7.1 Kafka 生产者与消费者

以下代码演示如何使用 Python 连接 Kafka，实现消息的生产和消费：

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import json
import time

class KafkaMessageQueue:
    """Kafka 消息队列封装类"""
    def __init__(self, bootstrap_servers: list):
        """ 初始化 Kafka 连接
        Args:
            bootstrap_servers: Kafka 集群地址列表
        """
        self.bootstrap_servers = bootstrap_servers
        self.producer = None
        self.consumer = None

    def create_producer(self, acks: str='all'):
        """ 创建生产者实例
        Args:
            acks: 确认级别 ('0', '1', 'all')
        """
        self.producer = KafkaProducer(
            bootstrap_servers=self.bootstrap_servers,
            acks=acks,
            key_serializer=lambda k: k.encode('utf-8') if k else None,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            retries=3,
            retry_backoff_ms=1000,
            batch_size=16384,
            linger_ms=10
        )
        return self.producer

    def send_message(self, topic: str, message: dict, key: str=None):
        """ 发送消息到指定 Topic
        Args:
            topic: 目标 Topic
            message: 消息内容（字典格式）
            key: 分区键（可选）
        """
        if not self.producer:
            self.create_producer()
        try:
            future = self.producer.send(topic, key=key, value=message)
            record_metadata = future.get(timeout=10)
            print(f"消息发送成功：Topic={record_metadata.topic}, Partition={record_metadata.partition}, Offset={record_metadata.offset}")
            return True
        except KafkaError as e:
            print(f"消息发送失败：{e}")
            return False

    def create_consumer(self, topic: str, group_id: str, auto_offset_reset: str='earliest'):
        """ 创建消费者实例
        Args:
            topic: 订阅的 Topic
            group_id: 消费者组 ID
            auto_offset_reset: 偏移量重置策略
        """
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=self.bootstrap_servers,
            group_id=group_id,
            auto_offset_reset=auto_offset_reset,
            enable_auto_commit=False,
            key_deserializer=lambda k: k.decode('utf-8') if k else None,
            value_deserializer=lambda v: json.loads(v.decode('utf-8')),
            max_poll_records=100,
            session_timeout_ms=30000
        )
        return self.consumer

    def consume_messages(self, process_func):
        """ 持续消费消息
        Args:
            process_func: 消息处理函数
        """
        if not self.consumer:
            raise ValueError("请先创建消费者实例")
        try:
            for message in self.consumer:
                try:
                    process_func(message.value)
                    self.consumer.commit()
                except Exception as e:
                    print(f"消息处理失败：{e}")
        except KeyboardInterrupt:
            print("消费者停止")
        finally:
            self.consumer.close()

# 使用示例
if __name__ == "__main__":
    mq = KafkaMessageQueue(['localhost:9092'])
    mq.send_message('orders', {'order_id': '12345', 'amount': 99.9})

    def process_order(msg):
        print(f"处理订单：{msg}")

    mq.create_consumer('orders', 'order-processor-group')
    mq.consume_messages(process_order)

上述代码封装了 Kafka 的生产者和消费者操作。生产者配置了 acks=all 确保消息可靠性，支持批量发送提升性能。消费者采用手动提交偏移量模式，确保消息处理成功后再提交。代码还包含了完善的错误处理和重试机制，适合生产环境使用。

7.2 RabbitMQ 生产者与消费者

以下代码演示如何使用 Python 连接 RabbitMQ，实现消息的生产和消费：

import pika
import json
from typing import Callable

class RabbitMQMessageQueue:
    """RabbitMQ 消息队列封装类"""
    def __init__(self, host: str, port: int=5672, username: str='guest', password: str='guest'):
        """ 初始化 RabbitMQ 连接
        Args:
            host: RabbitMQ 服务器地址
            port: RabbitMQ 端口
            username: 用户名
            password: 密码
        """
        self.connection_params = pika.ConnectionParameters(
            host=host,
            port=port,
            credentials=pika.PlainCredentials(username, password),
            heartbeat=600,
            blocked_connection_timeout=300
        )
        self.connection = None
        self.channel = None

    def connect(self):
        """建立连接"""
        self.connection = pika.BlockingConnection(self.connection_params)
        self.channel = self.connection.channel()
        return self.channel

    def declare_exchange(self, exchange_name: str, exchange_type: str='direct', durable: bool=True):
        """ 声明交换器
        Args:
            exchange_name: 交换器名称
            exchange_type: 交换器类型 (direct, topic, fanout, headers)
            durable: 是否持久化
        """
        if not self.channel:
            self.connect()
        self.channel.exchange_declare(
            exchange=exchange_name,
            exchange_type=exchange_type,
            durable=durable
        )

    def declare_queue(self, queue_name: str, durable: bool=True, arguments: dict=None):
        """ 声明队列
        Args:
            queue_name: 队列名称
            durable: 是否持久化
            arguments: 额外参数（如死信队列配置）
        """
        if not self.channel:
            self.connect()
        self.channel.queue_declare(
            queue=queue_name,
            durable=durable,
            arguments=arguments or {}
        )

    def bind_queue(self, queue_name: str, exchange_name: str, routing_key: str=''):
        """ 绑定队列到交换器
        Args:
            queue_name: 队列名称
            exchange_name: 交换器名称
            routing_key: 路由键
        """
        self.channel.queue_bind(
            queue=queue_name,
            exchange=exchange_name,
            routing_key=routing_key
        )

    def publish_message(self, exchange_name: str, message: dict, routing_key: str='', delivery_mode: int=2):
        """ 发布消息
        Args:
            exchange_name: 交换器名称
            message: 消息内容
            routing_key: 路由键
            delivery_mode: 投递模式 (1=非持久化，2=持久化)
        """
        if not self.channel:
            self.connect()
        properties = pika.BasicProperties(
            delivery_mode=delivery_mode,
            content_type='application/json'
        )
        self.channel.basic_publish(
            exchange=exchange_name,
            routing_key=routing_key,
            body=json.dumps(message),
            properties=properties
        )
        print(f"消息已发送：exchange={exchange_name}, routing_key={routing_key}")

    def consume_messages(self, queue_name: str, callback: Callable, prefetch_count: int=1):
        """ 消费消息
        Args:
            queue_name: 队列名称
            callback: 消息处理回调函数
            prefetch_count: 预取数量（用于流量控制）
        """
        if not self.channel:
            self.connect()
        self.channel.basic_qos(prefetch_count=prefetch_count)

        def on_message(ch, method, properties, body):
            try:
                message = json.loads(body)
                callback(message)
                ch.basic_ack(delivery_tag=method.delivery_tag)
            except Exception as e:
                print(f"消息处理失败：{e}")
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

        self.channel.basic_consume(
            queue=queue_name,
            on_message_callback=on_message,
            auto_ack=False
        )
        print(f"开始消费队列：{queue_name}")
        self.channel.start_consuming()

    def close(self):
        """关闭连接"""
        if self.connection and not self.connection.is_closed:
            self.connection.close()

# 使用示例
if __name__ == "__main__":
    mq = RabbitMQMessageQueue('localhost')
    mq.declare_exchange('order.exchange', 'topic')
    mq.declare_queue('order.created.queue')
    mq.bind_queue('order.created.queue', 'order.exchange', 'order.created')
    mq.publish_message('order.exchange', {'order_id': '12345', 'status': 'created'}, 'order.created')

    def process_order(msg):
        print(f"处理订单：{msg}")

    mq.consume_messages('order.created.queue', process_order)

上述代码封装了 RabbitMQ 的核心操作，包括交换器和队列的声明、绑定、消息发布和消费。代码采用手动确认模式确保消息可靠性，通过 prefetch_count 实现消费者流量控制。还支持死信队列配置和消息持久化，满足生产环境的可靠性要求。

7.3 Redis Stream 生产者与消费者

以下代码演示如何使用 Python 连接 Redis Stream，实现消息的生产和消费：

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

class RedisStreamMessageQueue:
    """Redis Stream 消息队列封装类"""
    def __init__(self, host: str='localhost', port: int=6379, db: int=0, password: Optional[str]=None):
        """ 初始化 Redis 连接
        Args:
            host: Redis 服务器地址
            port: Redis 端口
            db: 数据库编号
            password: 密码（可选）
        """
        self.client = redis.Redis(
            host=host,
            port=port,
            db=db,
            password=password,
            decode_responses=True
        )

    def add_message(self, stream_name: str, message: dict, maxlen: Optional[int]=None) -> str:
        """ 添加消息到 Stream
        Args:
            stream_name: Stream 名称
            message: 消息内容
            maxlen: 最大消息数量（可选，用于限制内存）
        Returns:
            消息 ID
        """
        fields = {}
        for key, value in message.items():
            fields[key] = json.dumps(value) if isinstance(value, (dict, list)) else str(value)

        if maxlen:
            message_id = self.client.xadd(stream_name, fields, maxlen=maxlen)
        else:
            message_id = self.client.xadd(stream_name, fields)
        print(f"消息已添加：stream={stream_name}, id={message_id}")
        return message_id

    def create_consumer_group(self, stream_name: str, group_name: str, start_id: str='0'):
        """ 创建消费者组
        Args:
            stream_name: Stream 名称
            group_name: 消费者组名称
            start_id: 起始消息 ID ('0' 表示从头开始，'$' 表示只消费新消息)
        """
        try:
            self.client.xgroup_create(stream_name, group_name, start_id)
            print(f"消费者组已创建：{group_name}")
        except redis.exceptions.ResponseError as e:
            if "BUSYGROUP" in str(e):
                print(f"消费者组已存在：{group_name}")
            else:
                raise

    def read_messages(self, stream_name: str, group_name: str, consumer_name: str, count: int=10, block: int=5000) -> list:
        """ 读取消息（消费者组模式）
        Args:
            stream_name: Stream 名称
            group_name: 消费者组名称
            consumer_name: 消费者名称
            count: 每次读取的消息数量
            block: 阻塞等待时间（毫秒）
        Returns:
            消息列表
        """
        messages = self.client.xreadgroup(group_name, consumer_name, {stream_name: '>'}, count=count, block=block)
        return messages or []

    def acknowledge_message(self, stream_name: str, group_name: str, message_id: str):
        """ 确认消息
        Args:
            stream_name: Stream 名称
            group_name: 消费者组名称
            message_id: 消息 ID
        """
        self.client.xack(stream_name, group_name, message_id)

    def get_pending_messages(self, stream_name: str, group_name: str) -> list:
        """ 获取待处理消息列表
        Args:
            stream_name: Stream 名称
            group_name: 消费者组名称
        Returns:
            待处理消息列表
        """
        pending = self.client.xpending_range(stream_name, group_name, '-', '+', count=100)
        return pending

    def claim_message(self, stream_name: str, group_name: str, consumer_name: str, message_ids: list, min_idle_time: int=60000):
        """ 认领超时消息（用于故障恢复）
        Args:
            stream_name: Stream 名称
            group_name: 消费者组名称
            consumer_name: 新消费者名称
            message_ids: 消息 ID 列表
            min_idle_time: 最小空闲时间（毫秒）
        """
        return self.client.xclaim(stream_name, group_name, consumer_name, min_idle_time, message_ids)

    def consume_continuous(self, stream_name: str, group_name: str, consumer_name: str, callback: Callable, count: int=10):
        """ 持续消费消息
        Args:
            stream_name: Stream 名称
            group_name: 消费者组名称
            consumer_name: 消费者名称
            callback: 消息处理回调函数
            count: 每次读取的消息数量
        """
        print(f"开始消费：stream={stream_name}, group={group_name}")
        while True:
            try:
                messages = self.read_messages(stream_name, group_name, consumer_name, count)
                for stream, msg_list in messages:
                    for message_id, fields in msg_list:
                        try:
                            parsed = {}
                            for key, value in fields.items():
                                try:
                                    parsed[key] = json.loads(value)
                                except (json.JSONDecodeError, TypeError):
                                    parsed[key] = value
                            callback(parsed)
                            self.acknowledge_message(stream_name, group_name, message_id)
                        except Exception as e:
                            print(f"消息处理失败：{e}")
            except redis.exceptions.ConnectionError:
                print("Redis 连接断开，尝试重连...")
                time.sleep(5)
            except KeyboardInterrupt:
                print("消费者停止")
                break

# 使用示例
if __name__ == "__main__":
    mq = RedisStreamMessageQueue('localhost')
    mq.create_consumer_group('orders', 'order-processor', '0')
    mq.add_message('orders', {'order_id': '12345', 'amount': 99.9})

    def process_order(msg):
        print(f"处理订单：{msg}")

    mq.consume_continuous('orders', 'order-processor', 'worker-1', process_order)

上述代码封装了 Redis Stream 的核心操作，包括消息添加、消费者组管理、消息读取和确认。代码支持阻塞读取、待处理消息查询、消息认领等高级功能。通过 maxlen 参数可以限制 Stream 长度，防止内存溢出。消费者组模式实现了消息的负载均衡和故障转移。

8. 总结

消息队列选型是分布式系统架构设计中的关键决策，直接影响系统的性能、可靠性和可维护性。本文从架构设计、消息模型、一致性保证、性能表现等多个维度，深入对比分析了 Kafka、RabbitMQ、Redis 三种主流消息队列方案。

核心要点回顾：

1. Kafka 以高吞吐量著称，采用磁盘顺序写和分区机制，单节点可达百万级消息/秒。适合日志收集、流处理等大数据场景，支持消息回溯和 Exactly-Once 语义。但部署复杂，需要 ZooKeeper 或 KRaft 协调服务，学习曲线较陡。
2. RabbitMQ 功能丰富，支持多种消息模型和路由策略，可靠性机制完善。适合订单处理、任务调度等复杂业务场景，支持延迟队列、死信队列等高级功能。内置管理界面，运维友好，但吞吐量相对较低。
3. Redis Stream 轻量高效，延迟最低，部署简单。适合实时通信、简单消息传递等场景，复用现有 Redis 基础设施。但功能有限，内存容量受限，不适合大规模消息存储。

选型建议：

• 大数据场景：选择 Kafka，充分利用其高吞吐量和消息回溯能力
• 复杂业务场景：选择 RabbitMQ，享受丰富的功能和可靠性保证
• 轻量级场景：选择 Redis Stream，简单高效，降低系统复杂度
• 混合架构：根据不同子场景选择合适的方案，组合使用

思考题：

1. 在你的业务场景中，消息丢失的容忍度如何？这如何影响你的选型决策？
2. 如果系统需要同时支持高吞吐量和复杂路由，你会如何设计混合架构？
3. 随着云原生技术的发展，托管消息服务（如 AWS SQS、阿里云 RocketMQ）是否是更好的选择？

参考资料

• Apache Kafka 官方文档
• RabbitMQ 官方文档
• Redis Stream 官方文档
• Kafka vs RabbitMQ vs Redis 性能对比研究
• 分布式消息队列最佳实践