Redis LPUSH RPOP List Queue Mock in Python

Implements a FIFO queue using Redis lists with LPUSH and RPOP commands, simulating task processing in Python.

Easy Python 3.9+ Aug 9, 2026 Caching & Redis 13 views 0 copies

Requires third-party packages — install first
pip install redis

Python code

18 lines
Python 3.9+
import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)
queue_key = 'task_queue'

# Push tasks onto the left side (LPUSH)
r.lpush(queue_key, 'task1')
r.lpush(queue_key, 'task2')
r.lpush(queue_key, 'task3')

# Mock processing: pop from the right side (RPOP) — FIFO order
while r.llen(queue_key) > 0:
    task = r.rpop(queue_key)
    print(f"Processing: {task.decode()}")
    time.sleep(0.1)

print("Queue empty")

Output

stdout
Processing: task1
Processing: task2
Processing: task3
Queue empty

How it works

This uses Redis lists as a simple message queue. LPUSH adds items to the left, and RPOP removes from the right, ensuring FIFO (First-In-First-Out) order. The loop continues while the list length is greater than zero, and each task is decoded from bytes. This pattern is common for lightweight task queues where you don't need advanced features like Pub/Sub or Streams.

Common mistakes

  • Forgetting to decode bytes returned by RPOP
  • Using time.sleep without a try/finally to handle interrupts
  • Assuming LPUSH and RPOP are atomic when popping in production (they are individually atomic but not for the whole loop)

Variations

  1. Use BRPOP instead of RPOP to block until a task becomes available
  2. Pop from the left with LPOP for LIFO (stack) behavior

Real-world use cases

  • Background job processing where tasks are added by web servers and consumed by workers.
  • Order processing in e-commerce systems where each order is queued and processed sequentially.
  • Log aggregation where messages are pushed to Redis and consumed by a logging pipeline.

Sponsored

Run locally

This sample needs third-party packages, so it cannot run in the browser IDE. Copy the code above, install the packages shown at the top, then run it in your own Python environment.

More from Caching & Redis

Related tutorials and quizzes for this topic.