How to Implement a Queue in Redis

Redis is a database that stores and processes data primarily in RAM. Because of this, Redis is often used to handle fast, short-lived data.

One common use case for Redis is caching. However, another powerful use is message queue processing, where Redis acts as a message broker like Apache Kafka or RabbitMQ.

That’s how a message broker works:

• Service 1 (e.g., a backend microservice) sends a message to the broker.
• Service 2 (or multiple services: Service 3, 4, 5, etc.) retrieves the message and processes it.

All messages are buffered, forming a message queue.

This allows multiple services to send messages simultaneously while others retrieve and process them over time.

Methods for Implementing a Queue in Redis

Copy link

Redis provides several built-in tools for implementing a message queue. Each method has its own advantages and limitations:

1. Pub/Sub (Publish/Subscribe)

• A service publishes a message to a queue.
• Only services subscribed to the queue can receive the message.
• If no subscribers are listening, the message is lost.

2. List (FIFO Queue: First In, First Out)

• The first message sent is the first message received.
• Each message is processed by only one subscriber.

3. Stream (Guaranteed Delivery)

• Works like Pub/Sub, but ensures message delivery.
• If no service is available to process the message, it remains in the queue until it is handled.

Creating a Redis Database in Hostman

Copy link

For this guide, we will use Hostman cloud databases.

1. Log in to Hostman and navigate to the Databases section.
2. Click Create database.
3. Select Redis as the database type.

4. Configure your database settings. For a test project, choose the minimal configuration to reduce costs.
5. Click Order to create the database.

The database will take a few moments to initialize and become active.

Once setup is complete, your Redis instance will be ready for message queue processing.

Queue Implementation

Copy link

Let's go through the process of implementing a queue in Redis step by step.

This guide uses a Hostman cloud server running Ubuntu 22.04.

Step 1. Update the system before configuring the server for a Python application:

sudo apt update
sudo apt upgrade

Step 2: Install Python. First, check if you already have Python:

python --version

The console output should look something like this:

Python 3.10.12

If Python is not installed, use the APT package manager to install it:

sudo apt install -y python3

The -y flag automatically answers "yes" to any prompts during installation.

Step 3: Install the Python virtual environment:

sudo apt install python3-venv -y

Step 4: Create a working directory for the Python project:

mkdir my-hostman-project

Navigate into the directory:

cd my-hostman-project

Step 5. Create a virtual Python environment in the working directory:

python -m venv venv

Now check the directory contents:

ls

If everything is set up correctly, a virtual environment folder should appear:

venv

Activate the environment:

source ./venv/bin/activate

Step 6: Install the Pip package manager

sudo apt install python3-pip -y

To verify the installation, check the Pip version:

pip -V

The console output should look something like this:

pip 22.0.2 from /usr/lib/python3/dist-packages/pip (python 3.10)

Step 7: Now install the Python module for working with Redis:

pip install redis

Later, we will import this module into the Python application.

Writing a Python Application

Copy link

Let's explore the basic ways to create a queue using Pub/Sub, List, and Stream.

Queue Using Pub/Sub

Copy link

In the working directory, create a handler file that will read the message queue:

sudo nano consumerPS.py

The Python code inside the file is as follows:

import redis
import time

connection = redis.Redis(
    host="IP",  # specify the Redis server's IP address
    password="PASSWORD",  # specify the Redis server's root password
    port=6379,  # standard port for connecting to Redis without SSL
    db=0,
    decode_responses=True  # automatically decodes Redis server responses into readable format
)

queue = connection.pubsub()  # create a Pub/Sub queue
queue.subscribe("channelFirst", "channelSecond")  # subscribe to the specified channels

# infinite loop for processing the message queue
while True:
    time.sleep(0.01)
    msg = queue.get_message()  # retrieve a message
    if msg:  # check if the message is empty
        if not isinstance(msg["data"], int):  # check the type of data in the "data" field (msg is a dictionary)
            print(msg["data"])  # print the message to the console

First, the script connects to the remote Redis server and then creates a Pub/Sub queue.

Note that when connecting to Redis, you must specify the remote host address and root password.  This example uses a non-SSL connection, so port 6379 is specified.

The queue subscribes to two channels: channelFirst and channelSecond.

Inside an infinite loop, the script periodically checks for new messages. If there is one, the console displays it. 

Now, create a sender file that will publish messages to the queue:

sudo nano producerPS.py

Its contents should be as follows:

import redis

# similar connection to the remote Redis server
connection = redis.Redis(
    host="IP",
    password="PASSWORD",
    port=6379,
    db=0,
    decode_responses=True
)  

connection.publish('channelFirst', 'This message was sent to the first channel')  # send a message to the first channel
connection.publish('channelSecond', 'This message was sent to the second channel')  # send a message to the second channel

First, the script connects to the remote Redis server in the same way as consumerPS.py. Then, two messages are sent over the open connection—one to the first channel and another to the second.

Now, we can execute the scripts to verify that the example works.

First, run the message handler in an open terminal window:

python consumerPS.py

Next, open a second terminal and activate the virtual environment:

source ./venv/bin/activate

Then, start the message sender:

python producerPS.py

As a result, the first terminal will display the following output:

This message was sent to the first channel

This message was sent to the second channel

Queue Using List

Copy link

Now, let's implement a similar queue but using the List entity.

First, create a handler file:

sudo nano consumerList.py

Write the following code in the file:

import redis
import random
import time

connection = redis.Redis(
    host="IP",
    password="PASSWORD",
    port=6379,
    db=0,
    decode_responses=True
)  

len = connection.llen("listQueue")  # get the size of the message queue list

# read messages from the list until the list size becomes zero
while connection.llen("listQueue") != 0:
    msg = connection.rpop("listQueue")  # read the message, which is a dictionary data type
    if msg:
        print(msg)  # print the message to the console

Note that in this example, we extract and remove the message "from the right" instead of "from the left." In other words, instead of using the lpop function, we use rpop.

Now, create the sender file:

sudo nano producerList.py

Its contents will be as follows:

import redis
import random

connection = redis.Redis(
    host="IP",
    password="PASSWORD",
    port=6379,
    db=0,
    decode_responses=True
)  

# send 3 messages at once
for i in range(0,3):
    connection.lpush("listQueue", "Message №" + str(random.randint(0, 100)))  # add a message with a unique number to the list

It is important to note that the messages are added to the list from right to left. For example, if 3 messages were sent:

Message №1
Message №2
Message №3

After that, the list will look like this:

[ Message №3, Message №2, Message №1 ]

Therefore, if the message handler code uses the rpop function, the messages will be processed in the order they were sent. If lpop is used, they will be processed in reverse order.

The same applies to sending messages using the rpush function instead of lpush.

Run the sender script to fill the message queue list:

python producerList.py

Then, process the messages:

python consumerList.py

The console should display output similar to this (only the message numbers will differ):

Message №94
Message №96
Message №24

Queue Using Stream

Copy link

Another useful tool for implementing a queue is "Streams." There are several basic commands for managing streams:

• XADD: Adds a new entry to the stream.
• XREAD: Reads one or more entries starting from a specified position and moving forward in time.
• XRANGE: Returns a range of entries between two provided record IDs.
• XLEN: Returns the length of the stream.

Create the message sender file:

sudo nano producerStream.py

The code inside should be:

import redis
import random

connection = redis.Redis(
    host="IP",
    password="PASSWORD",
    port=6379,
    db=0,
    decode_responses=True
)  

# send 3 messages at once
for i in range(0,3):
    connection.xadd("queueStream", { "data":"Message №" + str(random.randint(0, 100))})  # add a message with a unique number to the queue (dictionary type)

print("Queue length: " + str(connection.xlen("queueStream")))  # print the queue size to the console

In this example, we send 3 messages with a unique number to the stream via a for loop. After sending, the terminal will display the size of the stream.

Now, implement the message handler functionality:

sudo nano consumerStream.py

The code inside should be:

import redis
import random

connection = redis.Redis(
    host="IP",
    password="PASSWORD",
    port=6379,
    db=0,
    decode_responses=True
)  

len = connection.xlen("queueStream")  # get the length of the stream

if len > 0:
    messages = connection.xread(count=len, streams={"queueStream":0})  # get the entire list of messages in the stream

    # iterate over the list of messages
    for msg in messages:
        print(msg)  # print the message to the console

First, we extract the message queue from the stream and then process it sequentially in a for loop.

Run the written scripts in the following order:

1. Start the message sender:

python producerStream.py

The console should display:

Queue length: 3

2. Then, process the queue messages:

python consumerStream.py

The console output will look something like this:

['queueStream', [('1711712995031-0', {'data': 'Message №74'}), ('1711712995033-0', {'data': 'Message №54'})]]

From this output, you can notice that each message has a unique identifier automatically assigned by Redis.

However, this example has one drawback — each time we read the entire stream.

Let's improve the code in consumerStream.py so that each new script run reads only new messages from the stream:

import redis
import random

connection = redis.Redis(
    host="IP",
    password="PASSWORD",
    port=6379,
    db=0,
    decode_responses=True
)  

# create a Redis variable to store the ID of the last message (if this variable does not already exist)
if connection.get("last") == None:
    connection.set("last", 0)

len = connection.xlen("queueStream")  # get the length of the stream

if len > 0:
    messages = connection.xread(count=len, block=1000, streams={"queueStream":connection.get("last")})  # pass the last message ID as an argument (or 0)

    print(connection.get("last"))  # print the last message ID (or 0)

    # iterate over the list of new messages
    for msg in messages:
        print(msg)  # print the message to the console
        connection.set("last", msg[-1][-1][0])  # set the ID of the last read message as the value for the "last" variable

Now, each new request to Redis will print only fresh messages to the console.

Working with Redis streams is somewhat more complex than working with lists or subscribers. For a full understanding of streams when integrating this type of queue into your project, it's best to familiarize yourself with the official Redis examples.

Conclusion

Copy link

This guide demonstrated several basic ways to create a queue in Redis: Pub/Sub, List, and Stream.

The examples shown are minimal implementations that perform the logic of a message queue. A real project will require the logic to be more complex to meet developer criteria and solve specific tasks.

For instance, you can wrap the message queue functionality in classes and objects or implement it as a separate internal library for the project.

Each specific project will require further unique development of this implementation to solve its tasks.

To learn more about the Redis commands designed for working with different message queue tools, refer to the official Redis documentation:

• Pub/Sub
• List
• Stream