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


with instance() as client:
    client.rpush("arrival", b"b", b"a", b"b")
    assert client.lrange("arrival", 0, -1) == [b"b", b"a", b"b"]
    assert client.lindex("arrival", 1) == b"a"
    assert client.llen("arrival") == 3
    client.zadd("ranked", {"b": 10})
    client.zadd("ranked", {"a": 10})
    assert client.zadd("ranked", {"b": 20}) == 0
    assert client.zcard("ranked") == 2
    assert client.zrange("ranked", 0, -1, withscores=True) == [(b"a", 10.0), (b"b", 20.0)]
    client.zadd("ranked", {"b": 10})
    assert client.zrange("ranked", 0, -1) == [b"a", b"b"]
    assert client.zrange("ranked", 0, -1, desc=True) == [b"b", b"a"]
    print("lists retain duplicates and positions; sorted sets update unique members and order ties by bytes")

    assert client.lpop("arrival") == b"b"
    assert client.rpop("arrival") == b"b"
    assert client.lrange("arrival", 0, -1) == [b"a"]
    client.rpush("fifo", b"old", b"new")
    client.rpush("lifo", b"old", b"new")
    assert client.lpop("fifo") == b"old"
    assert client.rpop("lifo") == b"new"
    print("RPUSH with LPOP removes oldest first; RPUSH with RPOP removes newest first")

    client.zadd("times", {"at-10": 10, "at-20": 20, "at-30": 30})
    assert client.zrange("times", 0, 1) == [b"at-10", b"at-20"]
    assert client.zrange("times", 10, 20, byscore=True) == [b"at-10", b"at-20"]
    assert client.zrange("times", "(10", 20, byscore=True) == [b"at-20"]
    assert client.zremrangebyscore("times", "-inf", "(20") == 1
    assert client.zrange("times", 0, -1) == [b"at-20", b"at-30"]
    assert client.ttl("times") == -1
    client.rpush("window", b"old", b"middle", b"new")
    client.ltrim("window", -2, -1)
    assert client.lrange("window", 0, -1) == [b"middle", b"new"]
    print("rank selects positions; score selects values; score cleanup and list trimming are explicit")

    now = 100
    client.zadd("schedule", {"future-job": now + 50})
    assert client.zrange("schedule", "-inf", now, byscore=True) == []
    assert client.zpopmin("schedule") == [(b"future-job", 150.0)]
    assert client.exists("schedule") == 0
    print("ZPOPMIN removes the minimum immediately even when its score represents a future deadline")