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 concurrent.futures import ThreadPoolExecutor

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()


with instance() as client:
    client.rpush("fifo", b"a", b"b", b"c")
    assert [client.lpop("fifo") for _ in range(3)] == [b"a", b"b", b"c"]
    assert client.exists("fifo") == 0
    assert client.lpop("fifo") is None
    client.rpush("stack", b"a", b"b", b"c")
    assert [client.rpop("stack") for _ in range(3)] == [b"c", b"b", b"a"]
    client.lpush("left", b"a", b"b", b"c")
    assert client.lrange("left", 0, -1) == [b"c", b"b", b"a"]
    client.rpush("duplicates", b"same", b"same")
    assert client.llen("duplicates") == 2
    print("FIFO a,b,c; LIFO c,b,a; multi-value LPUSH reverses argument order; duplicates stay")

    def consume():
        with client.client() as connection:
            connection.client_setname("list-lab-consumer")
            return connection.blpop("channel", timeout=1)

    with ThreadPoolExecutor(max_workers=1) as worker:
        future = worker.submit(consume)
        deadline = time.monotonic() + 1
        while not any(
            entry["name"] == "list-lab-consumer" and "b" in entry["flags"]
            for entry in client.client_list()
        ):
            assert time.monotonic() < deadline, "consumer never blocked"
            time.sleep(0.01)
        client.rpush("channel", b"job:1")
        assert future.result(timeout=2) == (b"channel", b"job:1")
    assert client.blpop("empty", timeout=0.1) is None
    print("observed a blocked connection; RPUSH wakes it; empty BLPOP times out")

    client.rpush("ready", b"job:2", b"job:3")
    job = client.lmove("ready", "processing", "LEFT", "RIGHT")
    assert job == b"job:2"
    assert client.lrange("ready", 0, -1) == [b"job:3"]
    assert client.lrange("processing", 0, -1) == [b"job:2"]
    assert client.lmove("processing", "ready", "LEFT", "LEFT") == job
    assert client.lrange("ready", 0, -1) == [b"job:2", b"job:3"]
    assert client.lmove("ready", "processing", "LEFT", "RIGHT") == job
    assert client.lrem("processing", 1, job) == 1
    assert client.exists("processing") == 0
    print("LMOVE retains an in-flight job; explicit requeue and LREM acknowledgement work")

    client.set("wrong", b"not a list")
    try:
        client.lmove("ready", "wrong", "LEFT", "RIGHT")
    except redis.ResponseError as error:
        assert "WRONGTYPE" in str(error)
    else:
        raise AssertionError("LMOVE must reject a non-list destination")
    assert client.lrange("ready", 0, -1) == [b"job:3"]
    assert client.get("wrong") == b"not a list"
    client.rpush("recent", b"1", b"2", b"3", b"4")
    client.ltrim("recent", -2, -1)
    assert client.lrange("recent", 0, -1) == [b"3", b"4"]
    print("wrong-type move preserves source; LTRIM keeps only the requested window")