tokio-memq-python 0.1.3

Python bindings for tokio-memq

tokio-memq-python

tokio-memq-python is a high-performance, asynchronous in-memory message queue library for Python, powered by the robust Rust tokio-memq crate.

It bridges the gap between Python's ease of use and Rust's performance, providing a seamless asyncio interface for building high-throughput, low-latency messaging applications. Whether you are building a local event bus, a job queue, or a high-speed data pipeline, tokio-memq-python offers the speed and reliability you need.

Features

• 🚀 High Performance: Built on Rust's Tokio runtime, offering exceptional throughput and low latency.
• ⚡ Async/Await Native: Fully integrated with Python's asyncio for non-blocking I/O. Supports async for loops.
• � Consumer Groups: Built-in support for competing consumers pattern with automatic load balancing.
• � Partition Support: Scale horizontally with built-in partitioned topics and flexible routing strategies (RoundRobin, Hash, Random, Fixed).
• 🛠 Type Safety: Leverages Rust's strong type system to ensure stability and correctness.
• 🔧 Cross-Platform: Pre-compiled wheels available for Linux, Windows, and macOS (Intel & Apple Silicon).
• ⚙️ Configurable: Fine-tune queue depth, message TTL (Time-To-Live), and eviction policies (LRU).

� Use Cases

• High-Performance Event Bus: Decouple components in your Python application with minimal latency overhead.
• Async Job Processing: Efficiently distribute tasks to worker coroutines or threads using consumer groups.
• Data Streaming Pipelines: Handle high-velocity data ingestion and processing with backpressure support.
• Local Microservices Communication: Fast inter-process-like communication within a single application instance.
• Testing & Simulation: Simulate complex distributed messaging scenarios (like Kafka/RabbitMQ) locally without infrastructure overhead.

�📊 Benchmarks

Environment:

• OS: macOS
• Device: Apple M4 (24GB RAM)
• Python: 3.12

Results (Run with python3 examples/benchmark.py --count 10000):

Scenario	Metric	Result	Resource Usage (Avg CPU / Peak Mem)
Latency (1 Pub → 1 Sub)	Avg Latency	120.17 µs	N/A
	P99 Latency	174.55 µs
Throughput (1 Pub → 1 Sub)	Publish Rate	12,883 msg/s	148% / 37 MB
	Consume Rate	12,882 msg/s
Fan-Out (1 Pub → 5 Sub)	Publish Rate	6,647 msg/s	216% / 40 MB
	Total Consume Rate	28,306 msg/s
Fan-In (5 Pub → 1 Sub)	Publish Rate	32,974 msg/s	143% / 44 MB
	Consume Rate	13,662 msg/s

Note: Benchmarks include Python object serialization overhead. Raw Rust performance is significantly higher. Resource usage reflects the Python process overhead.

Installation

pip install tokio-memq-python

Requires Python 3.8 or later.

Examples

1. Basic Publish-Subscribe

A simple example demonstrating how to create a queue, publish messages, and consume them asynchronously.

import asyncio
from tokio_memq import MessageQueue

async def main():
    # Initialize the Message Queue
    mq = MessageQueue()
    
    # Create a publisher for "notifications"
    publisher = mq.publisher("notifications")
    
    # Create a subscriber for the same topic
    # This automatically creates the topic if it doesn't exist
    subscriber = await mq.subscriber("notifications")
    
    # Publish messages (any JSON-serializable data)
    await publisher.send({"type": "email", "to": "[e-mail]"})
    await publisher.send({"type": "sms", "to": "[phone]"})
    
    # Process messages using async iterator
    async for msg in subscriber:
        print(f"Processed notification: {msg}")
        # For this example, break after 2 messages
        if msg["type"] == "sms":
            break

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

2. Consumer Groups (Load Balancing)

Multiple consumers can join a group to share the workload. Messages are distributed among active consumers.

import asyncio
from tokio_memq import MessageQueue

async def worker(mq, topic, group_id, name):
    # Join a consumer group with "Earliest" mode to get existing messages
    sub = await mq.subscriber_group(topic, group_id, mode="Earliest")
    print(f"[{name}] Joined group '{group_id}' (ID: {sub.consumer_id})")
    
    async for msg in sub:
        print(f"[{name}] Processing: {msg}")

async def main():
    mq = MessageQueue()
    topic = "jobs"
    group = "workers"
    
    # Start two workers
    w1 = asyncio.create_task(worker(mq, topic, group, "Worker-1"))
    w2 = asyncio.create_task(worker(mq, topic, group, "Worker-2"))
    
    # Publish tasks
    pub = mq.publisher(topic)
    for i in range(10):
        await pub.send({"task_id": i})
        
    # Keep running...

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

3. Partitioned Topics with Hash Routing

This example shows how to use partitioned topics to distribute load while ensuring message ordering for specific keys (e.g., user IDs).

import asyncio
from tokio_memq import MessageQueue

async def main():
    mq = MessageQueue()
    topic = "user_activity"
    
    # Create a topic with 4 partitions for parallel processing
    await mq.create_partitioned_topic(topic, partition_count=4)
    
    # Configure "Hash" routing: messages with the same key go to the same partition
    await mq.set_partition_routing(topic, "Hash", key="message")
    
    pub = mq.publisher(topic)
    
    # Simulate user events
    # Events for 'user_123' will always land in the same partition
    await pub.send({"event": "login"}, key="user_123")
    await pub.send({"event": "view_item"}, key="user_123")
    
    # Events for 'user_456' will land in a different partition (likely)
    await pub.send({"event": "logout"}, key="user_456")
    
    # Consume from a specific partition (e.g., Partition 0)
    # Use mode="Earliest" to ensure we get messages published before we subscribed
    sub_p0 = await mq.subscribe_partition(topic, 0, mode="Earliest")
    
    print("Listening on Partition 0...")
    try:
        msg = await asyncio.wait_for(sub_p0.recv(), timeout=1.0)
        print(f"Partition 0 received: {msg}")
    except asyncio.TimeoutError:
        print("No messages on Partition 0 for this run.")

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

4. Advanced Configuration (TTL & Queue Limits)

Control memory usage and message lifecycle using TopicOptions.

import asyncio
from tokio_memq import MessageQueue, TopicOptions

async def main():
    mq = MessageQueue()
    
    # Configure topic options
    opts = TopicOptions()
    opts.max_messages = 100        # Limit queue to 100 messages
    opts.message_ttl_ms = 5000     # Messages expire after 5 seconds
    opts.lru_enabled = True        # Drop oldest messages when full
    
    # Create subscriber with custom options
    sub = await mq.subscriber_with_options("fast_logs", opts)
    pub = mq.publisher("fast_logs")
    
    # Publish a message
    await pub.send("This message will self-destruct in 5s")
    
    msg = await sub.recv()
    print(f"Got: {msg}")

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

More Examples

Check the examples/ directory for ready-to-run scripts:

• basic_usage.py: Complete pub/sub workflow.
• consumer_group.py: Load balancing with consumer groups.
• async_iteration.py: Pythonic usage with async for and send().
• partitioned_topic.py: Working with partitions and Hash routing.
• advanced_config.py: Using TTL and LRU eviction.
• performance_test.py: Simple throughput benchmark.
• benchmark.py: Comprehensive latency and throughput scenarios.

API Reference

MessageQueue

The central manager for topics, publishers, and subscribers.

• publisher(topic: str) -> Publisher: Get a publisher for a topic.
• subscriber(topic: str) -> Subscriber: Subscribe to a topic (auto-creates it).
• subscriber_with_options(topic: str, options: TopicOptions) -> Subscriber: Subscribe with specific configuration.
• subscriber_group(topic: str, consumer_id: str, mode: str = "LastOffset", offset: int = None) -> Subscriber: Join a consumer group.
  • consumer_id: Unique ID for the consumer group member.
  • mode: "Earliest", "Latest", "LastOffset", or "Offset".
• create_partitioned_topic(topic: str, partition_count: int, options: TopicOptions = None): Create a topic with n partitions.
• set_partition_routing(topic: str, strategy: str, key: str = None, fixed_id: int = None): Set routing logic.
  • strategy: "RoundRobin", "Random", "Hash", "Fixed".
  • key: Required for "Hash" routing (e.g., "message").
  • fixed_id: Required for "Fixed" routing (target partition ID).
• subscribe_partition(topic: str, partition_id: int, consumer_id: str = None, mode: str = "LastOffset", offset: int = None) -> Subscriber: Listen to a specific partition.
• list_partitioned_topics() -> List[str]: List active partitioned topics.
• delete_partitioned_topic(topic: str): Delete a topic.

Publisher

• send(data: Any, key: str = None): Send a message (alias for publish).
  • data: JSON-serializable payload (dict, list, str, int, etc.).
  • key: Routing key (used only if topic is partitioned with Hash routing).
• publish(data: Any, key: str = None): Send a message.
• topic() -> str: Get the topic name.

Subscriber

• recv() -> Any: Await the next message. Raises exception if connection is lost.
• try_recv() -> Any | None: Non-blocking receive. Returns None if no message available.
• current_offset() -> int: Get current offset.
• reset_offset(): Reset offset to 0.
• consumer_id -> str | None: (Property) Get the consumer ID if applicable.
• topic -> str: (Property) Get the topic name.
• Async Iterator: Supports async for msg in subscriber: syntax.

TopicOptions

• max_messages (int): Max messages per topic/partition.
• message_ttl_ms (int): Message lifetime in milliseconds.
• lru_enabled (bool): If True, drops oldest message when full; otherwise rejects new messages.
• partitions (int): (Read-only) Number of partitions if applicable.

License

This project is licensed under the MIT License.