Even more about Redis

Blocking operations: queues and process synchronization

BLPOP and BRPOP are the blocking versions of LPOP and RPOP. If the list is empty, the client waits until an item arrives or the timeout expires. They return a two-element reply [key, value], or None on timeout.

A FIFO queue is:

• producer: LPUSH (adds to the head of the list)

• consumer: BRPOP (removes from the tail of the list, blocking if empty)

common.pyfrom typing import Any

import redis

_CONN_PARS: Any = {
    "host": "localhost",
    "db": 0,
    "decode_responses": True,
    "socket_timeout": 1.0,
    "socket_connect_timeout": 1.0,
}


def connect(port: int = 6379) -> Any:
    return redis.Redis(**_CONN_PARS, port=port)


def cleanup(r: Any) -> None:
    if keys := r.keys("*"):
        r.delete(*keys)


def Title(title: str) -> None:
    print("\n===", title, "===\n")


def Line(tag: str, val: Any) -> None:
    print(f"{tag:16} -> {val}", flush=True)

blocking.pyimport multiprocessing as mp
import time

from common import cleanup, connect

with connect() as r:
    cleanup(r)

QUEUE = "queue:jobs"


# Producer pushes jobs slowly, then ends with a stop token
def producer():
    with connect() as r:
        time.sleep(0.5)  # give the consumer time to start blocking

        for i in range(1, 4):
            r.lpush(QUEUE, f"job-{i}")
            print(f"producer: pushed job-{i}", flush=True)
            time.sleep(0.3)

        r.lpush(QUEUE, "STOP")


# Consumer blocks until a job arrives
def consumer():
    with connect() as r:
        print("consumer: waiting on BRPOP ...", flush=True)
        while True:
            # blocks, timeout in seconds, 0 means block forever
            item = r.brpop(QUEUE, timeout=5)
            if item is None:
                print("consumer: timeout, nothing arrived")
                break

            _, job = item
            print("consumer: got", job, flush=True)
            if job == "STOP":
                break


process = mp.Process(target=producer)
process.start()

consumer()

process.join()

• Blocking operations can be also used to synchronize processes and pass messages between them.

• Do not put a blocking command inside MULTI/EXEC or a pipeline.

• A blocked client holds a connection from the pool.

Publish/subscribe: broadcast with no memory

• PUBLISH sends a message to every client currently subscribed to a channel, and returns the number of clients that received the message – zero is a valid answer.

• SUBSCRIBE listens to one channel, PSUBSCRIBE listens to a glob pattern.

• UNSUBSCRIBE and PUNSUBSCRIBE stop listening to channels.

• If no clients are subscribed to a channel, the message is discarded.

pubsub.pyimport time
from threading import Event, Thread
from typing import Any

from common import Line, Title, cleanup, connect

NEWS = "news"  # exact channel name
NEWS_STAR = "news.*"  # matches news.sports, news.weather, ...
# Note: "news.*" does NOT match plain "news" (the dot is required).


# listen for messages
def subscriber(ready: Event, stop: Event, received: list[Any]):
    # ignore_subscribe_messages=True hides the subscribe confirmation messages
    with connect().pubsub(ignore_subscribe_messages=True) as sub:
        sub.subscribe(NEWS)  # exact-channel subscription
        sub.psubscribe(NEWS_STAR)  # pattern subscription

        print("subscriber: listening on", NEWS, "and", NEWS_STAR, flush=True)
        ready.set()  # tell the main thread we are listening

        # polling loop
        while not stop.is_set():
            # a short timeout to keep the loop responsive
            message = sub.get_message(timeout=0.2)
            if message is None:
                continue

            match message["type"]:
                # delivered via subscribe()
                case "message":
                    Line("message", f"{message['channel']} -> {message['data']}")
                    received.append(message)

                # delivered via psubscribe(); the dict also has a "pattern" key
                case "pmessage":
                    Line(
                        "pattern",
                        f"{message['pattern']} matched {message['channel']}: {message['data']}",
                    )
                    received.append(message)

                # anything else
                case _:
                    Line("unknown", message)


Title("One publisher, one subscriber thread")

r = connect()
cleanup(r)

# thread-safe flags for signalling
ready = Event()
stop = Event()

received: list[Any] = []

thread = Thread(target=subscriber, args=(ready, stop, received))
thread.start()

# publisher:

ready.wait()  # wait for the subscriber to actually subscribe
time.sleep(0.2)  # allow SUBSCRIBE reach the server

Line("publish exact", r.publish(NEWS, "lecture starts"))
Line("publish pattern", r.publish("news.sports", "goal!"))
Line("publish ignored", r.publish("other", "nobody listens"))

# number of exact subscribers
Line("PUBSUB NUMSUB", r.pubsub_numsub(NEWS))

time.sleep(0.5)  # give the subscriber time to process

stop.set()  # signal the subscriber loop to end
thread.join()  # wait for the thread to finish

Line("messages received", len(received))

Title("Pub/Sub is send-and-forget")

# no subscribers on this channel
Line("publish to empty channel", r.publish("news.lost", "gone"))

# subscribing afterwards does not recover the gone message
with r.pubsub(ignore_subscribe_messages=True) as sub:
    sub.subscribe("news.lost")
    Line("late subscriber gets", sub.get_message(timeout=0.5))

An example of Redis replication

Start the Docker multicontainer with one master and two replica slaves:

compose.yaml# Redis replication demo: one image (redis:7-alpine), three servers.
# A master accepts writes; two replicas copy everything from it, read-only.
# "x-" keys are ignored by Compose; this is just a reusable YAML anchor (&redis)
# holding settings shared by all services.
x-redis: &redis
  image: redis:7-alpine # official image, no custom Dockerfile needed

services:
  redis-master:
    <<: *redis # merge in the shared settings (the image)
    # --protected-mode "no": using no passwords for a local demo.
    # Do not do this in production.
    command: ["redis-server", "--protected-mode", "no"]
    ports: ["6380:6379"]

  redis-replica1:
    <<: *redis
    # --replicaof <host> <port>:
    # connect to the master, do a full sync, then keep receiving its writes.
    command: ["redis-server", "--protected-mode", "no", "--replicaof", "redis-master", "6379"]
    ports: ["6381:6379"]
    # This only orders startup; replicas retry until the master answers.
    depends_on: [redis-master]

  redis-replica2:
    <<: *redis
    command: ["redis-server", "--protected-mode", "no", "--replicaof", "redis-master", "6379"]
    ports: ["6382:6379"]
    depends_on: [redis-master]

Check the health and replication status (in the terminal):

bashPORTS="6380 6381 6382"

for port in $PORTS; do redis-cli -p $port ping; done
for port in $PORTS; do echo $port; redis-cli -p $port info replication | grep role; done

Try to write to the replicas, see that the slaves are read-only:

bashfor port in $PORTS; do echo $port; redis-cli -p $port set foo bar; done

Read from the replicas, see that the data is replicated from the master:

bashfor port in $PORTS; do echo $port; redis-cli -p $port get foo; done

An example of Redis sharding

Redis Cluster is a sharded Redis setup with multiple masters and replicas.

The key is hashed using CRC16, the result modulo 16384 (CRC16(key) mod 16384) determines the hash slot. Each master is responsible for a range of hash slots.

Replicas are assigned to masters for redundancy.

Start the Docker multicontainer with three masters and one replica for each master:

compose-cluster.yaml# Redis Cluster demo: 6 nodes = 3 masters (shards) + 3 replicas.

x-cluster-node: &cluster-node
  image: redis:7-alpine
  # Host networking, Linux only.
  network_mode: host

# --cluster-enabled yes        turn on cluster mode (slots, redirects)
# --cluster-config-file        where the node stores its view of the cluster
# --cluster-node-timeout 5000  ms without a reply before a node is "failing"
# --appendonly no              no persistence needed for a demo
services:
  node1:
    <<: *cluster-node
    command: ["redis-server", "--cluster-enabled", "yes", "--cluster-config-file", "nodes.conf", "--cluster-node-timeout", "5000", "--appendonly", "no", "--port", "7001"]
  node2:
    <<: *cluster-node
    command: ["redis-server", "--cluster-enabled", "yes", "--cluster-config-file", "nodes.conf", "--cluster-node-timeout", "5000", "--appendonly", "no", "--port", "7002"]
  node3:
    <<: *cluster-node
    command: ["redis-server", "--cluster-enabled", "yes", "--cluster-config-file", "nodes.conf", "--cluster-node-timeout", "5000", "--appendonly", "no", "--port", "7003"]
  node4:
    <<: *cluster-node
    command: ["redis-server", "--cluster-enabled", "yes", "--cluster-config-file", "nodes.conf", "--cluster-node-timeout", "5000", "--appendonly", "no", "--port", "7004"]
  node5:
    <<: *cluster-node
    command: ["redis-server", "--cluster-enabled", "yes", "--cluster-config-file", "nodes.conf", "--cluster-node-timeout", "5000", "--appendonly", "no", "--port", "7005"]
  node6:
    <<: *cluster-node
    command: ["redis-server", "--cluster-enabled", "yes", "--cluster-config-file", "nodes.conf", "--cluster-node-timeout", "5000", "--appendonly", "no", "--port", "7006"]

  # One-shot job: joins the six empty nodes into a cluster, then exits.
  # --cluster-replicas 1 -> 3 masters, each with 1 replica; 
  # the 16384 slots are divided between the masters. 
  # --cluster-yes accepts the proposed layout.
  cluster-init:
    <<: *cluster-node
    depends_on: [node1, node2, node3, node4, node5, node6]
    restart: "no"
    # Nodes may not be accepting connections yet, so wait for them first.
    entrypoint: ["/bin/sh", "-c"]
    command:
      - |
        for p in 7001 7002 7003 7004 7005 7006; do
          until redis-cli -p $$p ping >/dev/null 2>&1; do sleep 0.5; done
        done
        redis-cli --cluster create \
          127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 \
          127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006 \
          --cluster-replicas 1 --cluster-yes

Check the health and replication status:

bashPORTS="7001 7002 7003 7004 7005 7006"
for port in $PORTS; do redis-cli -p $port ping; done
for port in $PORTS; do echo $port; redis-cli -p $port info replication | grep role; done
for port in $PORTS; do echo $port; redis-cli -p $port cluster nodes | grep myself; done
for port in $PORTS; do echo $port; redis-cli -p $port cluster info | grep cluster_state; done

Write data to the cluster, see that the keys are distributed among the masters:

bashredis-cli -p 7001 set foo bar # error: MOVED
redis-cli -p 7003 set foo bar # OK
redis-cli -p 7001 set foa bar # OK
redis-cli -p 7001 set fob bar # error: MOVED

With redis-cli -c (enable cluster mode, follow redirects), the client can write to any node in the cluster (even a replica), which will redirect to the responsible master:

bashredis-cli -c -p 7001 set foo bar
redis-cli -c -p 7001 set foa bar
redis-cli -c -p 7001 set fob bar

Read the keys back, see that they are distributed among the masters:

bashredis-cli -p 7001 get foo # error: MOVED
redis-cli -p 7003 get foo # OK
redis-cli -p 7001 get foa # OK
redis-cli -p 7002 get foa # error: MOVED
redis-cli -p 7001 get fob # error: MOVED
redis-cli -p 7003 get fob # OK

With -c, the client can read from any node in the cluster:

bashredis-cli -c -p 7001 get foo
redis-cli -c -p 7003 get foo
redis-cli -c -p 7001 get foa
redis-cli -c -p 7002 get foa
redis-cli -c -p 7001 get fob
redis-cli -c -p 7003 get fob