Redis 发布订阅广播消息但不保留回放

redis_lab=$(mktemp -d)
curl --fail --location --output "$redis_lab/redis.tar.gz" https://download.redis.io/releases/redis-8.10.1.tar.gz
tar -xzf "$redis_lab/redis.tar.gz" -C "$redis_lab"
make -C "$redis_lab/redis-8.10.1" build redis -j2 MALLOC=libc
export REDIS_SERVER="$redis_lab/redis-8.10.1/src/redis-server"
python3 -m venv "$redis_lab/venv"
"$redis_lab/venv/bin/python" -m pip install redis==8.1.0

from contextlib import contextmanager
import os
from pathlib import Path
import subprocess
from tempfile import TemporaryDirectory
import time

import redis
from redis.backoff import NoBackoff
from redis.retry import Retry


@contextmanager
def instance():
    with TemporaryDirectory(prefix="redis-lab-", dir="/tmp") as directory:
        endpoint = str(Path(directory) / "redis.sock")
        process = subprocess.Popen([
            os.environ["REDIS_SERVER"], "--port", "0",
            "--unixsocket", endpoint, "--unixsocketperm", "700",
            "--save", "", "--appendonly", "no", "--dir", directory,
            "--maxmemory-policy", "noeviction",
        ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
        client = redis.Redis(unix_socket_path=endpoint, decode_responses=False,
                             socket_timeout=2, retry=Retry(NoBackoff(), 0))
        try:
            deadline = time.monotonic() + 5
            while True:
                assert process.poll() is None, "Redis exited during startup"
                try:
                    if client.ping():
                        break
                except redis.ConnectionError:
                    if time.monotonic() >= deadline:
                        raise
                    time.sleep(0.02)
            yield client
        finally:
            client.close()
            if process.poll() is None:
                process.terminate()
            try:
                _, errors = process.communicate(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.communicate()
                raise
            assert process.returncode == 0, errors.decode()


def receive(subscription):
    deadline = time.monotonic() + 2
    while time.monotonic() < deadline:
        message = subscription.get_message(timeout=0.1)
        if message is not None:
            return message
    raise AssertionError("subscription response timed out")


with instance() as client:
    assert client.publish("events", b"before-subscribe") == 0
    with client.pubsub() as first, client.pubsub() as second:
        for subscription in (first, second):
            subscription.subscribe("events")
            response = receive(subscription)
            assert response["type"] == "subscribe"
            assert response["channel"] == b"events" and response["data"] == 1

        for value in (b"one", b"two"):
            assert client.publish("events", value) == 2
        for subscription in (first, second):
            messages = [receive(subscription) for _ in range(2)]
            assert [message["type"] for message in messages] == ["message"] * 2
            assert [message["data"] for message in messages] == [b"one", b"two"]
        print("before subscription: 0; two listeners receive one,two in order")

        second.unsubscribe("events")
        response = receive(second)
        assert response["type"] == "unsubscribe" and response["data"] == 0
        assert client.publish("events", b"missed") == 1
        assert receive(first)["data"] == b"missed"
        second.subscribe("events")
        assert receive(second)["type"] == "subscribe"
        assert client.publish("events", b"after-resubscribe") == 2
        for subscription in (first, second):
            assert receive(subscription)["data"] == b"after-resubscribe"
        print("resubscription receives the new marker, not the missed message")

        first.psubscribe("event*")
        response = receive(first)
        assert response["type"] == "psubscribe" and response["data"] == 2
        assert client.publish("events", b"overlap") == 3
        messages = [receive(first) for _ in range(2)]
        assert {message["type"] for message in messages} == {"message", "pmessage"}
        assert all(message["data"] == b"overlap" for message in messages)
        assert receive(second)["data"] == b"overlap"
        print("two clients, three matching deliveries; one client receives two copies")