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 threading import Barrier

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:
    payload = b"\x00\xff" + "\u8336".encode("utf-8")
    client.set("binary", payload)
    assert client.get("binary") == payload
    assert client.strlen("binary") == 5
    assert client.getrange("binary", 2, 2) == b"\xe8"
    assert client.getrange("binary", 2, 4).decode("utf-8") == "\u8336"
    print("binary round trip: 5 bytes; GETRANGE addresses bytes")

    start = Barrier(4)

    def increment(_):
        start.wait(timeout=5)
        return [client.incr("counter") for _ in range(50)]

    with ThreadPoolExecutor(max_workers=4) as workers:
        batches = list(workers.map(increment, range(4)))
    values = sorted(value for batch in batches for value in batch)
    assert values == list(range(1, 201))
    assert client.get("counter") == b"200"
    print("4 workers: 200 INCR replies, exactly the integers 1 through 200")

    for key, value in [("invalid", b"3.5"),
                       ("overflow", b"9223372036854775807")]:
        client.set(key, value)
        try:
            client.incr(key)
        except redis.ResponseError:
            pass
        else:
            raise AssertionError("INCR must reject this value")
        assert client.get(key) == value
    print("invalid integer and signed 64-bit overflow leave values unchanged")

    client.set("document", b'{"score": 1}')
    assert client.type("document") == b"string"
    try:
        client.hget("document", "score")
    except redis.ResponseError as error:
        assert "WRONGTYPE" in str(error)
    else:
        raise AssertionError("JSON bytes must not become a Redis hash")
    client.hset("object", mapping={"score": 1})
    client.set("object", b"replacement")
    assert client.type("object") == b"string"
    print("JSON is still a string; SET can replace a hash with a string")

    client.set("lease", b"A", px=60000)
    deadline = client.pexpiretime("lease")
    client.set("lease", b"B", keepttl=True)
    assert client.pexpiretime("lease") == deadline
    client.set("lease", b"C")
    assert client.ttl("lease") == -1
    print("SET PX assigns expiry; KEEPTTL keeps it; plain SET removes it")