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

比较等价的工作

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

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

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

计入应用实际支付的成本

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

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

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

坏掉的样子

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

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

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

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

假想四命令梯形图说明批次逻辑等待；精确 2,048 命令夹具的等待计数与计时、原子性、重试及生产边界分开。
假想梯形图解释逻辑等待。夹具批次计数是算术，不是实测 RTT、延迟或容量；历史速度测量仍未核实。

快速验证

原文报告的环境是 macOS、Python 3.14.4、Redis 8.10.1 和 redis-py 8.1.0；本轮查阅的证据中未找到匹配的执行日志，也没有重新运行。需要 POSIX、C 编译器、make、curl、tar 和 Python 3.10 或更新版本。必须启用断言，不能使用 -O 或 PYTHONOPTIMIZE，否则检查及其中必需的调用都会被跳过。

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" 运行，并逐条确认安装命令成功；下载需要联网。Redis 使用 /tmp 下的私有 Unix socket，关闭 TCP、RDB 定时快照和 AOF。清理流程尝试停止自己的进程并移除运行目录；下载、构建及虚拟环境所在目录仍会保留。代码不访问默认本机 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、持久化开销、连接故障和集群扩展。部署前的基准应分别覆盖这些维度，不用一个快速测试代替完整容量判断。

来源