在 Python 的异步编程生态（Asyncio）中，我们经常需要在不同的协程或模块之间传递消息。虽然标准库提供的asyncio.Queue简单好用，但在面对高并发、多消费者组负载均衡、或者需要更复杂的路由策略时，它往往显得力不从心。而部署 Redis、RabbitMQ 等外部消息队列对于单机或嵌入式应用来说又显得过于“重”了。
今天为大家介绍一个兼顾极致性能与丰富功能的轻量级解决方案：tokio-memq-python。
什么是 tokio-memq-python？
tokio-memq-python是一个基于 Rust 语言编写的高性能内存消息队列库，并通过 PyO3 提供了原生的 Python 绑定。它的核心由 Rust 强大的异步运行时Tokio驱动，旨在弥合 Python 的易用性与 Rust 的高性能之间的鸿沟。
项目地址：https://github.com/weiwangfds/tokio-memq-python
PyPI 地址：https://pypi.org/project/tokio-memq-python/
Rust Crate 地址：https://crates.io/crates/tokio-memq
核心特性
为什么选择它而不是标准的asyncio.Queue？
- 🚀 极致性能：底层逻辑完全由 Rust 实现，利用 Tokio 的高效调度，在处理高吞吐量消息时表现优异。
- ⚡ 原生 Asyncio 支持：完美集成 Python 的
async/await语法，支持async for异步迭代消费，使用体验与原生 Python 库无异。 - 👥 消费者组 (Consumer Groups)：支持类似 Kafka 的消费者组模式。多个消费者可以加入同一个组，消息会自动在组内成员间进行负载均衡（Competing Consumers），非常适合构建工作流池。
- 🧩 分区主题 (Partitioned Topics)：支持主题分区，可以将消息通过轮询、哈希或随机策略路由到不同分区，进一步提升并发吞吐能力。
- 🛠 类型安全与稳定：得益于 Rust 的内存安全保证，在高并发场景下更加稳定可靠。
- 🔧 跨平台：提供预编译的 Wheel 包，支持 Linux, Windows, macOS (含 Apple Silicon)。
性能表现
在 Apple M4 (24GB RAM) 设备上的基准测试结果令人印象深刻：
| 场景 | 指标 | 结果 |
|---|---|---|
| 延迟 (1 Pub → 1 Sub) | 平均延迟 | 120.17 µs |
| P99 延迟 | 174.55 µs | |
| 高扇入 (5 Pub → 1 Sub) | 发布速率 | 32,974 msg/s |
| 高扇出 (1 Pub → 5 Sub) | 总消费速率 | 28,306 msg/s |
注：以上数据包含了 Python 对象序列化的开销，底层 Rust 的处理能力远高于此。
快速上手
1. 安装
需要 Python 3.8+ 环境：
pipinstalltokio-memq-python2. 基础：发布与订阅
使用起来非常简单，就像使用本地对象一样：
importasynciofromtokio_memqimportMessageQueueasyncdefmain():# 初始化消息队列实例mq=MessageQueue()# 创建发布者publisher=mq.publisher("user_events")# 创建订阅者subscriber=awaitmq.subscriber("user_events")# 发送消息（支持任意可 JSON 序列化的数据）awaitpublisher.send({"event":"login","user_id":1001})awaitpublisher.send({"event":"click","btn":"signup"})# 异步迭代消费消息asyncformsginsubscriber:print(f"收到事件:{msg}")ifmsg["event"]=="click":breakif__name__=="__main__":asyncio.run(main())3. 进阶：使用消费者组进行负载均衡
这是tokio-memq-python相比标准 Queue 最强大的功能之一。你可以启动多个消费者 worker，它们会自动分摊任务。
importasynciofromtokio_memqimportMessageQueueasyncdefworker(mq,group_id,worker_name):# 加入消费者组 "image_processors"# mode="Earliest" 表示如果之前有积压消息，从最早的开始消费sub=awaitmq.subscriber_group("images",group_id,mode="Earliest")print(f"[{worker_name}] 已就绪")asyncformsginsub:print(f"[{worker_name}] 处理图片:{msg['file']}")# 模拟处理耗时awaitasyncio.sleep(0.1)asyncdefmain():mq=MessageQueue()publisher=mq.publisher("images")# 启动 3 个消费者 workergroup_id="image_processors"workers=[asyncio.create_task(worker(mq,group_id,f"Worker-{i}"))foriinrange(3)]# 发布任务foriinrange(10):awaitpublisher.send({"file":f"photo_{i}.jpg"})awaitasyncio.sleep(2)# 等待处理完成if__name__=="__main__":asyncio.run(main())适用场景
- 高性能事件总线：在单进程应用内解耦各个模块。
- 异步任务处理：替代 Celery/Redis 处理轻量级的本地异步任务分发。
- 数据流管道：处理高频数据采集、缓冲和分发。
- 本地微服务通信：在单一应用实例中模拟微服务架构的通信模式。
- 测试与仿真：在开发环境中模拟 Kafka/RabbitMQ 的行为，无需启动重型容器。
结语
tokio-memq-python是一个专注于性能与开发体验的现代化消息队列库。如果你正在寻找内存的消息队列，但又不想引入外部依赖的解决方案，它绝对值得一试。
欢迎大家 Star 支持！🌟
GitHub: https://github.com/weiwangfds/tokio-memq-python
