turbomq 0.1.10

TurboMQ - Message Queue System

TurboMQ

TurboMQ is a simple message queue system. I hope it is fast enough to merit the name. In our tests, it could produce and consume millions of messages per second, but we leave the final judgment to developers who use the library. Note that it is currently experimental, and there may be dramatic changes in both functionality and protocols.

Why was TurboMQ developed?

First, I want to explain why a new message queue system was developed. There are many message queue systems available, and some of them are popular and stable, such as RabbitMQ and ZMQ. The most important reason behind this implementation is that most message queue systems are designed to handle backend processing, such as distributing jobs between nodes to process huge amounts of data or completing the remaining part of a business transaction. Certainly, TurboMQ can be used to distribute work between nodes. However, it was originally designed to support millions of producers and consumers working with millions of queues and topics.

The closest system in terms of queue functionality is Redis. It has a remarkable I/O mechanism for handling network connections. However, one instance can utilize only one core. Do we really want to use only one core out of, for example, eight available cores? Or do we want to configure clustering inside one machine just to use all available cores?

ZMQ is a good library. It is fast, stable, and useful for many purposes. Nonetheless, there is a serious problem in topic-based PUB-SUB queues: consumers (subscribers) have to be connected before producers (missing message problem solver), otherwise messages can be lost.

Technical Information

TurboMQ is a Python module. To avoid GIL problems, it is developed using pure C and Cython. It uses its own event loop system. The benefit is that it is a real multi-threaded event loop and can exploit all available cores. The drawback is that it does not support Windows. Is that all the bad news? No, kqueue has not been implemented yet, and it uses slow POSIX poll on BSD systems. Is there any other good news? Yes, Windows and kqueue support are planned.

Installation

Installation is easy. The package can be installed with uv:

$ uv pip install turbomq

Alternatively, you can download it or clone it directly from GitHub and install it locally:

$ git clone https://github.com/turbomq/engine.git
$ cd engine
$ uv sync
$ uv pip install .

For development, build the Cython extension in place:

$ uv run python setup.py build_ext --inplace

Usage

To use TurboMQ, just import it and run the server. The following code runs a server for 10 minutes.

from turbomq import TurboEngine
import time

# You can pass the thread count as a second parameter.
# Otherwise, it automatically selects 4 threads per core.
e = TurboEngine('tcp://127.0.0.1:33444')

e.run()
# "run" method will not block the main thread.
# So you need to wait or run your own loop.
time.sleep(10.0 * 60)

# The "stop" method just shuts TCP sockets down.
e.stop()

# After destroy, all resources will be freed.
# You cannot use this instance anymore.
e.destroy()

This code sends a message to the server and receives it again.

from turbomq import TurboClient

# Connect to the server.
c = TurboClient('tcp://127.0.0.1:33444')

# Create a mirror queue on the client side.
q = c.get_queue('test')

# Both the topic key and data are mandatory in push.
q.push('hello', 'turbo')

# In pop, you need to specify a timeout.
# This waits two seconds. If the timeout is exceeded, it returns None.
print(q.pop('hello', 2))

It is possible to produce messages on the client side and consume them on the server side, and vice versa:

Server code:

from turbomq import TurboEngine
import time

e = TurboEngine('tcp://127.0.0.1:33444')

# Make the engine ready to serve remote clients.
e.run()

# Create a queue on the server side.
q = e.get_queue('test')

# Wait to consume client commands.
while True:
    # Wait one second for a message.
    m = q.pop('hello', 1)
    if m is not None:
        print('Client says: ' + m.content.decode('utf-8'))
        # Put a new message for the client in another topic.
        q.push('hello', 'Hi Client.')
        break

# Wait for 2.0 seconds.
time.sleep(2.0)

# Clean up and shut the engine down.
e.stop()
e.destroy()

Client code:

from turbomq import TurboClient

# Connect to the server.
c = TurboClient('tcp://127.0.0.1:33444')

# Create a mirror queue on the client side.
q = c.get_queue('test')

# Send the message to the server.
q.push('hello', 'Hi Server.')

# Wait one second to pop the server message.
m = q.pop('hello', 1)
print('Server says: ' + m.content.decode('utf-8'))

Future

First, I will fix bugs to make the protocol and engine stable. The second step is to implement client libraries for other programming languages. Then we will extend the functionality and features.