流水线减少的是发送 Redis 命令之间的等待，实际收益必须放到具体负载里测量。远程同步循环可能主要等待网络延迟，本地循环也可能主要消耗在客户端处理、进程调度和套接字读写。两种情形都不能直接推出“任何环境都会快几倍”的结论，更不能用一次本机数字替线上系统承诺容量。

比较等价的工作

对于 N 条命令、每批 B 条，简单同步模型把约 N 次请求响应等待减少为向上取整的 N/B 次批次等待。但命令执行、响应字节、编码和解析并没有消失；一批数据仍可能跨越多次套接字写入和多个网络包。因此模型中的等待次数，不是抓包数量，也不是对真实链路 RTT 的测量。

改变 B 时，要固定命令、键分布、负载大小、连接数、客户端版本和持久化设置。否则看起来更快，可能只是做了更少工作、关掉了持久性，或者增加了并发客户端。若要测普通流水线，应显式使用 transaction=False，因为 redis-py 默认流水线还会加入 MULTI/EXEC。

下面批量一使用普通同步客户端循环，不是每次只装一条命令的流水线包装器。这样比较回答的是“应用当前请求响应循环能否从批处理受益”，而不是测出 Redis 的最大可持续吞吐，更不是剥离客户端后得到服务器的固有命令成本。

计入应用实际支付的成本

计时应覆盖命令准备，直到最后一个结果收齐。忽略构建请求或解析响应，会低估大批次给应用带来的成本。示例把缓冲、执行和结果收集都放进计时区间，而把重置数据与正确性断言放在外面；两种路径遵守相同的计时边界，才方便比较。

每个配置先预热，再变换测试顺序并重复测量，同时报告波动。这里的三次正式运行只是一个容易复现的小型比较，不是严谨的容量研究。笔记本上的调度和其它工作仍会影响结果，单次中位数不能直接升级成服务等级目标。

吞吐量也不是单条请求延迟。总耗时除以命令数，得到的是平均完成成本，不是百分之九十五或九十九请求的延迟分布。等待凑满一批，可能让最早到达的请求在执行前就排队；本例立即构造批次，没有测量真实到达节奏带来的积攒延迟。

坏掉的样子

巨型批次可能提高每秒命令数，却同时带来难以接受的停顿和内存增长。应限制条数、请求与响应字节，以及等待成批的时间；这些上限应来自代表性负载，而不是不分命令类型就照搬几百或几千条。

如果继续增大批量已经没收益，要检查客户端 CPU、响应解码、服务端命令成本、带宽和竞争负载。流水线能够通过分摊读写开销提高实际服务吞吐，但并不消除所有瓶颈。分片是可能的架构选择之一，不是曲线变平之后唯一合理的答案。

计速度也必须检查失败。丢掉回复或把错误当成功，可以轻易制造漂亮但无意义的速率。下方关闭网络自动重试，并核对每一个递增结果的完整序列。真实断连可能留下不确定的执行范围；从最后一条已知回应之后重发，对递增也不是天然安全的恢复方式。

测试负载只应留在它自己的实例中。代码会反复重置并递增一个计数器，若把目标改成共享生产库，操作风险就发生了变化。这里使用本地 Unix 套接字，也有意排除了 TCP、TLS 和远程网络路径，所以读数不能冒充跨机房链路测量。

快速验证

原有实验记录使用 macOS、Python 3.14.4、Redis 8.10.1 和 redis-py 8.1.0；本轮审阅没有重新执行服务实验，下面的结果说明保留原有观察范围。准备 POSIX 环境、C 编译器、make、curl、tar 与 Python 3.10 或更新版本。安装步骤仅用 libc 构建 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

在同一终端用 "$redis_lab/venv/bin/python" 运行下方代码。它在 /tmp 启动只用私有 Unix 套接字的 Redis，关闭 TCP 和持久化，结束时仅停止自己的进程并清理运行目录。下载需要联网，实验不会连接已有 Redis 实例。

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


from statistics import median


TOTAL = 2048
SIZES = (1, 8, 64, 256)


def measure(client, size):
    client.set("bench:counter", 0)
    started = time.perf_counter()
    responses = []
    if size == 1:
        for _ in range(TOTAL):
            responses.append(client.incr("bench:counter"))
    else:
        for offset in range(0, TOTAL, size):
            with client.pipeline(transaction=False) as batch:
                for _ in range(min(size, TOTAL - offset)):
                    batch.incr("bench:counter")
                responses.extend(batch.execute())
    elapsed = time.perf_counter() - started
    assert responses == list(range(1, TOTAL + 1))
    assert client.get("bench:counter") == str(TOTAL).encode()
    return elapsed


with instance() as client:
    for size in SIZES:
        measure(client, size)
    measurements = {size: [] for size in SIZES}
    for repetition in range(3):
        order = SIZES[repetition:] + SIZES[:repetition]
        for size in order:
            measurements[size].append(measure(client, size))
    for size in SIZES:
        samples = measurements[size]
        duration = median(samples)
        waits = (TOTAL + size - 1) // size
        print(f"batch={size:3d} waits={waits:4d} median_ms={duration * 1000:.2f} "
              f"commands_per_second={TOTAL / duration:.0f} "
              f"range_ms={min(samples) * 1000:.2f}-{max(samples) * 1000:.2f}")
    print("all replies and final counters checked; no speedup threshold is asserted")

每种配置都在同一个计数器上执行二千零四十八次递增，并复用连接。批量一、八、六十四和二百五十六，对应的逻辑阻塞调用次数分别是二千零四十八、二百五十六、三十二和八。这些是应用等待次数，不是捕获到的数据包数；每轮都会核对完整返回序列和最后的存储值。

原有记录的中位完成时间依次为 35.24、9.40、5.61 和 5.00 毫秒。批量越大，后续收益逐步缩小，展示了该本地负载上的边际变化。脚本打印的是运行者自己的测量及实际范围，刻意没有设置“必须达到某种加速比”的断言，环境不同就应得到不同数字。

四种配置各先预热一次，再以轮换顺序完成三次正式运行。每秒命令数用二千零四十八除以计时结果计算，不能解读为不同并发、负载大小和网络条件下远程应用的预测性能。

实验验证的是等价工作都确实成功，并测量本地整批完成时间，没有测生产到达率、尾延迟、TCP 或 TLS、持久化开销、连接故障和集群扩展。部署前的基准应分别覆盖这些维度，不用一个快速测试代替完整容量判断。

来源