Skip to content
Published on

消息队列与异步入门：Queue·Kafka·RabbitMQ·Redis·asyncio 一览

🇰🇷KO🇺🇸EN🇯🇵JA🇨🇳ZH
필사 모드
分享
Authors

引言 — 为什么需要队列

服务一旦变大，"不必现在立刻处理"的工作就会越来越多。注册后发送欢迎邮件、给上传的视频编码、支付后汇总结算。如果在处理请求的过程中同步地做这些事，用户就得干等。所以我们把这些工作放进队列，留到后面慢慢处理。

消息队列正是把这个想法做成了系统。生产者(producer)把消息放进去，消费者(consumer)取出来处理。队列夹在中间，让生产和消费解耦(decouple)、吸收速度差(buffering)，即便消费者挂掉，消息也还在，可以重新处理(reliability)。

问题在于，"消息队列"这一个词底下，聚集着性格相当不同的工具。Kafka、RabbitMQ、Redis 都被称为队列，但设计哲学各不相同。本文将它们逐一对照，最后再梳理常和队列混淆的 Python asyncio。

如果想亲眼看看这些概念是怎么运作的，可以在本站新做的消息队列游乐场里交互式地可视化体验。

从最朴素的 FIFO 队列说起

最简单的队列是先进先出(FIFO)的数据结构。先进去的先出来。编程语言的标准库里也自带这种队列。

from collections import deque

q = deque()
q.append("job-1")   # enqueue
q.append("job-2")
print(q.popleft())  # dequeue -> job-1
print(q.popleft())  # dequeue -> job-2

这种队列的核心性质有三条：顺序被保留、一条消息只发给一个消费者、一旦取出就消失。概念很清楚，局限也很清楚。这个 deque 只存在于一个进程的内存里。进程一旦挂掉，队列也随之消失，也无法和另一台服务器上的消费者共享。

分布式消息队列系统正是为了突破这个局限而存在的。它们把队列搬到网络之外，存进磁盘，让多个生产者和消费者都能挂接上来。只不过各自的做法不尽相同。

Kafka — 无法擦除的日志

Kafka 其实算不上传统意义上的"队列"。Kafka 是一份只增日志(append-only log)。它不是把消息从队列里取出后销毁，而是不断把消息追加到日志文件末尾，消费者只记住"自己读到哪里了"。

核心概念梳理如下。

  • 主题(topic)与分区(partition)：主题是消息的分类，每个主题又被切分成若干分区。分区是并行的单位。
  • 偏移量(offset)：每个分区内消息的序号。消费者会保存自己的偏移量，如果愿意，可以回退到过去的偏移量做重放(replay)。
  • 消费者组(consumer group)：同一组内的消费者会分摊各个分区。一个分区在组内只会分配给一个消费者。所以增加消费者能提升吞吐量，但一旦消费者数量超过分区数量，多出来的消费者就会闲置。
  • 仅分区内保证顺序：顺序只在分区内部有保证，整个主题不保证全局顺序。

画成图是这样的：

  主题: orders
  ┌─────────────────────────────────────────┐
  │ 分区 0: [o0][o1][o2][o3] ...  <- 消费者 A
  │ 分区 1: [o0][o1][o2] ...      <- 消费者 B
  │ 分区 2: [o0][o1][o2][o3][o4]  <- 消费者 C
  └─────────────────────────────────────────┘
     每个消费者都记着自己的偏移量

Kafka 强大的原因在于日志会留存下来。消息不会在被消费的瞬间就消失，所以新的消费者晚些时候才接入，也能从头重新读起。这很适合事件溯源、流处理，以及多个系统各自消费同一批事件的架构。相应地，你要始终记住顺序是按分区来保证的，并且"相同的键必须发到同一个分区"顺序才能成立。

RabbitMQ — 聪明的路由

RabbitMQ 是实现了 AMQP 协议的传统消息代理。如果说 Kafka 是一份日志，那 RabbitMQ 就是一间聪明的邮局。关键在于生产者并不会直接把消息放进队列。生产者把消息发给交换机(exchange)，再由交换机按规则把消息路由到队列。

路由方式由交换机的类型决定。

  • direct：发送给路由键完全匹配的队列。
  • fanout：忽略路由键，复制一份发给所有绑定的队列。属于广播。
  • topic：把路由键当作模式来匹配。比如用 order.* 或 order.# 这样的模式做部分匹配。

生产者、交换机、绑定、队列、消费者之间的关系是这样的：

  生产者 --> [交换机] --绑定(路由键)--> [队列] --> 消费者
                 │
                 ├── direct  : 键完全匹配
                 ├── fanout  : 全部复制
                 └── topic   : 模式匹配 (order.*, order.#)

RabbitMQ 的另一个核心是确认应答(ack)。消费者处理完消息并发送 ack 后，队列才会把这条消息删除。如果消费者在处理途中挂掉、没能发出 ack，代理会把这条消息重新投递给另一个消费者。这样就能避免"处理中丢失"。

与 Kafka 的决定性区别在于，消息被消费后就会消失。RabbitMQ 不适合把过去的消息倒回去重放。取而代之的是，它擅长精细地控制消息流：复杂路由、任务分发、优先级队列、延迟队列。

Redis — 轻量而多才多艺的三种方式

Redis 原本是内存数据存储，但也被广泛用于消息传递。有趣的是，Redis 提供了三种截然不同的消息传递方式。

1. 基于列表的队列(LPUSH / BRPOP)。 最简单的任务队列。一边往列表里推，另一边从里面取。BRPOP 是阻塞方式，没有消息时消费者就会等待。

# 生产者
LPUSH tasks "job-1"
LPUSH tasks "job-2"

# 消费者 (最多等待 5 秒,直到有消息到来)
BRPOP tasks 5

2. Pub/Sub(发布/订阅)。 一旦向频道发布，当下所有正在订阅的消费者都会立刻收到。但这是即发即忘(fire-and-forget)。如果没有订阅者，消息就直接消失，不会被存储。适合"只发给此刻正在听的人"这类场景，比如实时通知。

# 订阅者
SUBSCRIBE news

# 发布者 (发给所有正在订阅的人,不会被存储)
PUBLISH news "hello"

3. Streams(流)。 Redis Streams 是列表的简单性和 Kafka 的持久性、消费者组之间的折中方案。消息像日志一样堆积，可以分给消费者组消费，也支持处理确认(ack)。当你想要类似 Kafka 的模式、但规模又用不上 Kafka 时，它很趁手。

# 追加到流
XADD mystream * event "signup" user "alice"

# 创建消费者组后读取
XGROUP CREATE mystream g1 0
XREADGROUP GROUP g1 consumer1 COUNT 1 STREAMS mystream >

Redis 的魅力在于，可以把队列轻松叠加在"你本来就当缓存用的那个 Redis"上。不过这三种方式的性质各不相同，选择时要先想清楚：能不能接受丢失(Pub/Sub)，还是不能(列表、Streams)。

什么时候用什么 — 比较表

把到目前为止的内容整理成一张表，如下所示。

项目FIFO 队列KafkaRabbitMQRedis
模型内存数据结构分区的只增日志AMQP 代理列表/Pub-Sub/Streams
消费后的消息消失保留(可重放)消失(ack 后)因方式而异
顺序保证全局仅分区内仅队列内列表、Streams 会保留
路由无键 → 分区交换机(强大)简单
优势简单大规模流、可重放复杂路由、任务分发轻量、多用途
典型用途进程内缓冲事件溯源、日志流水线微服务任务队列缓存兼队列、实时通知

简单的选型指南是这样的：

  • 如果只是一个进程内的临时缓冲，标准库队列就够了。
  • 如果需要长期保留事件、并让多个消费者重放和重新处理，选 Kafka。
  • 如果需要复杂路由、任务分发，以及可靠的基于 ack 的处理，选 RabbitMQ。
  • 如果已经在用 Redis，只需要轻量队列或实时通知，Redis 很实用。

异步和队列不是一回事 — Python asyncio

这里要澄清一个常见的误解。"异步处理"和"消息队列"经常一起出现，但并不是同一件事。消息队列是横跨进程和服务器的基础设施，而 asyncio 是在单个进程内处理并发的编程模型。

Python asyncio 的核心是单线程事件循环。它不增加线程，而是由一个线程在多个任务之间来回切换处理。诀窍在于 I/O 等待时间。在等待网络响应或磁盘的时候，先把这个任务暂停(await)，趁这个空当去推进另一个任务。

import asyncio

async def fetch(name, delay):
    print(f"start {name}")
    await asyncio.sleep(delay)   # 这段时间里其他协程会执行
    print(f"done {name}")
    return name

async def main():
    # 同时推进三个任务(接近取最大值,而不是把等待时间加总)
    results = await asyncio.gather(
        fetch("a", 2),
        fetch("b", 1),
        fetch("c", 3),
    )
    print(results)

asyncio.run(main())

这里再梳理几个术语。

  • 协程(coroutine)：用 async def 定义的函数。可以中途暂停、之后恢复。
  • await：意味着"在这里等待，但把控制权让给事件循环，让它趁这段时间做别的事"。
  • gather：把多个协程一起调度，让它们同时推进。

最重要的区分是并发(concurrency)与并行(parallelism)。asyncio 给的是并发。多个任务看起来在交叠推进，但真正在同一时刻执行的始终只有一个线程。相对地，并行是在多个 CPU 核心上真正同时执行。

这个区别在实务中意味着什么很明确。asyncio 擅长I/O 密集型工作(等待网络、磁盘、数据库)。等待的时间越多，收益越大。但它对CPU 密集型工作(繁重的计算)没有帮助。计算过程中没有可以让出的空隙，事件循环会被卡住。CPU 密集型工作需要多进程或专门的工作进程。

队列与 asyncio 会一起使用

两者不同，并不意味着彼此无关。恰恰相反，它们经常一起使用。典型的结构是这样的。Web 服务器接到请求后，不会自己去做繁重的工作，而是把消息放进消息队列。另一个独立的工作进程从队列里取出任务处理，而在这个工作进程内部，又用 asyncio 同时处理多条消息。

  请求 --> Web 服务器 --(消息)--> [队列: Kafka/RabbitMQ/Redis]
                                      |
                                      v
                                  工作进程
                                  └ 用 asyncio 同时处理多条消息

也就是说，队列是"把工作挪到以后、挪给另一个进程"的那一层，而 asyncio 是"在一个进程内高效利用等待时间"的那一层。层次不同，所以两者不是竞争关系，而是互补关系。

常见的坑

最后来看看实务中经常踩到的几个点。

  • "恰好一次"(exactly-once)的幻觉：大多数队列默认保证的是"至少一次(at-least-once)"。也就是说，同一条消息有可能收到两次。因此消费者要设计成幂等(idempotent)的，同一条消息被处理多次也要得到相同的结果。
  • 对顺序过度信任：Kafka 的顺序是按分区保证的。如果需要全局顺序，要么只用一个分区，要么谨慎设计键，而这都会牺牲并行性。
  • 对 Pub/Sub 抱有持久化的期待：Redis Pub/Sub 不会存储消息。错过的消息就是没了。如果丢失是个问题，应该用列表、Streams，或者别的代理。
  • 在 asyncio 里做阻塞调用：在事件循环里调用同步的阻塞函数(比如普通的 time.sleep，或者阻塞式的数据库驱动)，会让整个循环卡住。要用支持 async 的库，或者把它交给单独的线程去做。
  • 无视死信(dead letter)：对一条不断失败的消息无限重试，会把队列堵住。更安全的做法是另外设一个死信队列，专门收集失败的消息。

结语

消息队列不是单一的概念，而是一个光谱。从进程内简单的 FIFO 队列出发，一路延伸到无法擦除的日志(Kafka)、聪明的路由(RabbitMQ)、轻量而多才多艺的工具(Redis)。而与它们气质不同、却经常一起使用的 asyncio，则是在单个进程内高效处理等待时间的并发模型。

关键不在于"哪个是正确的"，而在于"哪个适合这个问题"。需要重放，用 Kafka；需要路由，用 RabbitMQ；需要轻量，用 Redis；需要进程内并发，用 asyncio。把这套对应关系记在脑子里，大多数选择就会变得容易。

如果想亲自动手体验这些概念，可以到消息队列游乐场看看每种方式的行为有何不同。

参考资料

Discuss on Twitter • View on GitHub