在 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？

1. 🚀 极致性能：底层逻辑完全由 Rust 实现，利用 Tokio 的高效调度，在处理高吞吐量消息时表现优异。
2. ⚡ 原生 Asyncio 支持：完美集成 Python 的 async/await 语法，支持 async for 异步迭代消费，使用体验与原生 Python 库无异。
3. 👥 消费者组 (Consumer Groups)：支持类似 Kafka 的消费者组模式。多个消费者可以加入同一个组，消息会自动在组内成员间进行负载均衡（Competing Consumers），非常适合构建工作流池。
4. 🧩 分区主题 (Partitioned Topics)：支持主题分区，可以将消息通过轮询、哈希或随机策略路由到不同分区，进一步提升并发吞吐能力。
5. 🛠 类型安全与稳定：得益于 Rust 的内存安全保证，在高并发场景下更加稳定可靠。
6. 🔧 跨平台：提供预编译的 Wheel 包，支持 Linux, Windows, macOS (含 Apple Silicon)。

性能表现

在 Apple M4 (24GB RAM) 设备上的基准测试结果令人印象深刻：

注：以上数据包含了 Python 对象序列化的开销，底层 Rust 的处理能力远高于此。

快速上手

1. 安装

需要 Python 3.8+ 环境：

pip install tokio-memq-python

2. 基础：发布与订阅

使用起来非常简单，就像使用本地对象一样：

import asyncio
from tokio_memq import MessageQueue

async def main():
    # 初始化消息队列实例
    mq = MessageQueue()
    
    # 创建发布者
    publisher = mq.publisher("user_events")
    
    # 创建订阅者
    subscriber = await mq.subscriber("user_events")
    
    # 发送消息（支持任意可 JSON 序列化的数据）
    await publisher.send({"event": "login", "user_id": 1001})
    await publisher.send({"event": "click", "btn": "signup"})
    
    # 异步迭代消费消息
    async for msg in subscriber:
        print(f"收到事件: {msg}")
        if msg["event"] == "click":
            break

if __name__ == "__main__":
    asyncio.run(main())

3. 进阶：使用消费者组进行负载均衡

这是 tokio-memq-python 相比标准 Queue 最强大的功能之一。你可以启动多个消费者 worker，它们会自动分摊任务。

import asyncio
from tokio_memq import MessageQueue

async def worker(mq, group_id, worker_name):
    # 加入消费者组 "image_processors"
    # mode="Earliest" 表示如果之前有积压消息，从最早的开始消费
    sub = await mq.subscriber_group("images", group_id, mode="Earliest")
    print(f"[{worker_name}] 已就绪")
    
    async for msg in sub:
        print(f"[{worker_name}] 处理图片: {msg['file']}")
        # 模拟处理耗时
        await asyncio.sleep(0.1)

async def main():
    mq = MessageQueue()
    publisher = mq.publisher("images")
    
    # 启动 3 个消费者 worker
    group_id = "image_processors"
    workers = [
        asyncio.create_task(worker(mq, group_id, f"Worker-{i}"))
        for i in range(3)
    ]
    
    # 发布任务
    for i in range(10):
        await publisher.send({"file": f"photo_{i}.jpg"})
    
    await asyncio.sleep(2) # 等待处理完成

if __name__ == "__main__":
    asyncio.run(main())

适用场景

• 高性能事件总线：在单进程应用内解耦各个模块。
• 异步任务处理：替代 Celery/Redis 处理轻量级的本地异步任务分发。
• 数据流管道：处理高频数据采集、缓冲和分发。
• 本地微服务通信：在单一应用实例中模拟微服务架构的通信模式。
• 测试与仿真：在开发环境中模拟 Kafka/RabbitMQ 的行为，无需启动重型容器。

结语

tokio-memq-python 是一个专注于性能与开发体验的现代化消息队列库。如果你正在寻找内存的消息队列，但又不想引入外部依赖的解决方案，它绝对值得一试。

欢迎大家 Star 支持！🌟

GitHub: https://github.com/weiwangfds/tokio-memq-python