★ FEATURED ARTICLE 简介一份基于 Python 与 Redis 的任务队列示例项目借助 RQ 库实现异步任务的生产与消费适合刚接触分布式任务队列的开发者快速上手。整个压缩包仅 3KB共包含 5 个文件类型覆盖 Python 源码、编译生成的 pyc、Markdown 文档和 gitignore 配置文件其中 py 文件承载核心逻辑md 文件记录环境准备与运行说明工程配置也一并给出每个文件的功能定位都很清晰。项目演示了从 Redis 服务启动、rqworker 任务监听到运行任务投递脚本的完整流程并在说明文档中提供了基于豆瓣镜像源和 Docker 的环境准备命令包含 Redis Server 的拉取与启动方式便于本地快速复现实验环境。已有 385 人学习这份资源虽然体量很小但通过这个完整可运行的示例程序读者既能理解 RQ 队列的调用链路与 worker 工作方式也能对照 pyc 文件初步认识 Python 字节码形态是一份轻量级的任务队列入门参考。1. 任务队列到底是什么一个自备 Redis 的 Python Demo 能帮你少走多少弯路很多 Python 开发者第一次接触任务队列都是从 Celery 或者 RQ 开始但这两者都自带了一整套调度、序列化和 Worker 管理逻辑遇到问题时常感觉像在调一个黑匣子。这个名为 python-redis-queue-sample 的 Demo 反其道而行它只依赖一个 Redis 实例和标准库级的 redis-py 客户端用几十行代码把「生产者 - 队列 - 消费者」这条链路完整跑通。说白了它解决的是「我有个耗时操作不想同步阻塞接口想丢到后台慢慢跑」这个最朴素的需求尤其适合正在做课程设计、内部工具或小型服务的从业者——能看清每一步在干什么也能随时改成自己的业务逻辑。自备 Redis 环境即可复现接下来我会从选型开始带你把这份 Demo 拆开揉碎。2. 选型与最小实现从 List 到 Stream先跑通一个能用的队列2.1 为什么用 Redis 做队列而不是数据库轮询或消息中间件把任务塞进 MySQL 表、再定时扫描状态字段是很多人第一个想到的方案但它在任务量稍大时会出现两个问题轮询间隔难以取舍扫太频繁浪费数据库连接扫太慢任务延迟高另外并发消费时 UPDATE 状态容易产生竞争条件。而 RabbitMQ、Kafka 这类消息中间件功能强但部署和运维成本高对于 Demo 或中小型项目来说明显过重。Redis 的 List 结构天然支持LPUSHBRPOP这种「阻塞弹出」模式生产者往队列左侧推任务消费者阻塞在队列右侧等待任务到达即刻弹出。关键在于BRPOP是阻塞调用不会像轮询那样空转消耗 CPU配合 Redis 本身的单线程事件模型多个消费者可以同时阻塞等待由 Redis 保证每个元素只被一个消费者取走。这个 Demo 的代码结构非常直白核心就是两个 Python 脚本一个模拟生产任务一个模拟消费处理。我实际跑通后觉得它选 List 而不是 Stream 作为默认存储是合理的——List 操作简单、直观对「先入先出 阻塞消费」这个场景完全够用。Stream 虽然支持消费者组、消息确认这些高级特性但理解成本和代码复杂度都上了一个台阶适合在 Demo 基础上做二次演进。2.2 最小可运行版本生产者与消费者的完整代码先看生产者端它做的事情很简单构造任务字典序列化成 JSON然后推入 Redis 队列。import json import redis # 连接池配置block 模式下连接会长期占用必须给足连接数 pool redis.ConnectionPool( host127.0.0.1, port6379, db0, max_connections30, decode_responsesTrue, socket_connect_timeout5, socket_timeoutNone, retry_on_timeoutTrue ) r redis.Redis(connection_poolpool) def publish_task(task_type: str, payload: dict) - str: task { task_id: f{task_type}-{int(time.time() * 1000)}, type: task_type, payload: payload, created_at: time.time() } # LPUSH 推入队列左侧消费者从右侧 BRPOP 取出 r.lpush(task_queue, json.dumps(task)) return task[task_id] # 模拟产生 10 个图片处理任务 if __name__ __main__: for i in range(10): task_id publish_task( image_resize, {image_path: f/data/images/sample_{i}.jpg, width: 800} ) print(fpublished: {task_id})这里有几个参数值得展开说明。decode_responsesTrue让 redis-py 自动把字节串解码成字符串省得每次取数据都要自己decode()socket_timeoutNone是刻意为之因为消费者使用阻塞读取时不能有读超时否则BRPOP会被打断retry_on_timeoutTrue则是在网络抖动导致读超时时让客户端自动重试获取连接而不是直接抛异常。max_connections设 30 是因为每个消费者线程都会长期占用一个连接如果后续要开 20 个消费者线程连接数不够就会排队等连接表现为消费延迟异常升高。再看消费端这是理解整个 Demo 的关键。BRPOP的返回值是一个[key, value]列表需要解包处理。import json import time import redis pool redis.ConnectionPool( host127.0.0.1, port6379, db0, max_connections5, decode_responsesTrue, socket_connect_timeout5, socket_timeoutNone, retry_on_timeoutTrue ) r redis.Redis(connection_poolpool) QUEUE_KEY task_queue BLOCK_TIMEOUT 5 # 阻塞等待 5 秒超时返回 None def process_task(task_data: dict): task_type task_data.get(type) payload task_data.get(payload, {}) if task_type image_resize: image_path payload.get(image_path) width payload.get(width) # 模拟耗时操作真实场景这里会调用 Pillow 或 OpenCV print(fresizing {image_path} to width{width}) time.sleep(0.5) else: print(funknown task type: {task_type}) def run_worker(): print(worker started, waiting for tasks...) while True: try: # BRPOP 会阻塞 BLOCK_TIMEOUT 秒超时返回 None然后继续循环 result r.brpop(QUEUE_KEY, timeoutBLOCK_TIMEOUT) if result is None: continue queue_name, raw_data result task_data json.loads(raw_data) process_task(task_data) except json.JSONDecodeError as e: # 数据格式损坏时不退出记录后继续 print(fjson decode error: {e}, raw: {raw_data}) except Exception as e: print(ftask process error: {e}, retry after 1s) time.sleep(1) if __name__ __main__: run_worker()这个消费端包含了一个初学者最容易忽略的设计BLOCK_TIMEOUT引入了 5 秒的空转周期。BRPOP阻塞 5 秒后如果没有任务会返回None代码里continue继续下一次阻塞这看似无害但实际上每 5 秒会产生一次空轮询。在低任务量场景下可以接受但如果你希望任务一到就立刻处理BLOCK_TIMEOUT应该设大一些比如 30 秒或 60 秒这样 Redis 连接重新建立的频率更低。异常处理方面json.JSONDecodeError单独捕获非常必要。实际生产环境中队列里偶尔会出现半截数据或脏字符如果在这里直接崩溃退出整个 Worker 就停了。而通用Exception捕获后的time.sleep(1)是为了防止任务本身抛出致命异常导致死循环疯狂打印日志这个退避机制虽然简单但能救你一次。2.3 跑通后的验证方法与预期结果启动顺序有讲究一定要先启动消费者再运行生产者。如果顺序反了任务会被堆在 Redis 里消费者启动后一次性全部消费看不出实时效果。# 终端 1启动消费者 python worker.py # 终端 2启动生产者 python producer.py正常情况下消费者终端会持续输出resizing /data/images/sample_0.jpg to width800之类的日志每 0.5 秒处理一条直到 10 个任务全部完成。此时在 Redis 里执行LLEN task_queue应该返回 0。如果你想观察阻塞状态下的连接行为可以在 Producer 运行期间打开另一个终端执行redis-cli CLIENT LIST会看到消费者连接处于cmdbrpop状态。这个细节能帮你确认阻塞确实生效了而不是空轮询。3. 生产端细节序列化、连接池与优先级任务怎么落地3.1 任务序列化方案JSON 以外还有哪些选择各有什么坑Demo 里默认使用 JSON 序列化这个选择在日常开发中是安全的但有几个边界情况你必须提前想清楚。JSON 无法序列化datetime、Decimal、自定义对象如果你把一个包含datetime的任务字典直接json.dumps会在生产端直接抛TypeError。常见做法是自定义一个DefaultEncoder来兜底import json from datetime import datetime, date from decimal import Decimal class TaskEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, (datetime, date)): return obj.isoformat() if isinstance(obj, Decimal): return float(obj) # bytes 类型转成 base64 字符串避免反序列化失败 if isinstance(obj, bytes): return obj.decode(utf-8, errorsreplace) return super().default(obj) # 使用方式 task_json json.dumps(task, clsTaskEncoder, ensure_asciiFalse)参数ensure_asciiFalse值得强调。如果任务 payload 里包含中文路径或中文描述默认的ensure_asciiTrue会把所有非 ASCII 字符转成\uXXXX转义序列虽然能正常反序列化回来但数据在 Redis 里可读性极差排查问题时用redis-cli LRANGE看原始任务会非常痛苦。更复杂的数据可以考虑pickle它支持 Python 任意对象但存在两个致命问题版本兼容性差不同 Python 小版本之间 pickle 协议可能不兼容反序列化任意 pickle 字节码有代码执行风险。除非是内部可信环境且你清楚风险否则我不推荐在生产队列里用 pickle。MessagePack 是另一个选项序列化后体积比 JSON 小 30% 左右但需要额外依赖对于这个小 Demo 来说收益不明显。3.2 连接池配置误区每个 Worker 一个连接还是一个进程一个连接池很多人第一次写 redis 队列时会在每个 Worker 函数里直接redis.Redis(...)新建连接这会导致两个后果每次新建连接都有 TCP 握手开销高并发时 Redis 服务器的连接数会暴涨很快打满maxclients配置。正确做法是整个进程共享一个连接池每个线程从池里借连接用完归还。Demo 中的连接池配置有一个隐形的坑socket_timeoutNone配合连接池复用如果 Redis 服务器异常重启池里的旧连接可能处于半死状态。redis-py 在从池里取连接时会做一个health_check但前提是你在初始化时设置了health_check_interval。建议显式加上pool redis.ConnectionPool( host127.0.0.1, port6379, db0, max_connections50, decode_responsesTrue, socket_connect_timeout5, socket_timeoutNone, retry_on_timeoutTrue, health_check_interval30 # 每 30 秒检查一次连接健康状态 )health_check_interval30的意思是当从连接池里取出的连接空闲超过 30 秒时redis-py 会先发送一个PING检查连接是否存活失败则丢弃重建。这个参数在长生命周期 Worker 里是必配的否则你可能会遇到「任务处理越来越慢最后卡死不动」的玄学问题实际是连接已经断了但客户端还傻等。多个消费者进程的场景下我一般建议每个进程独立创建连接池而不是进程间共享。因为连接池不是线程安全的容器多个进程共享一个池实例容易踩到内部状态竞争的雷。3.3 优先级任务用两个队列模拟优先级的简单方案真实业务里经常遇到「普通任务可以等但用户主动触发的任务必须马上执行」的需求。Redis List 本身不支持优先级但可以用多队列 按序读取来模拟。这个思路在 Demo 基础上只需要改动消费端的阻塞逻辑HIGH_QUEUE task_queue_high NORMAL_QUEUE task_queue def run_worker_with_priority(): while True: # 先阻塞读高优先级队列超时后读普通队列 result r.brpop([HIGH_QUEUE, NORMAL_QUEUE], timeout5) if result is None: continue queue_name, raw_data result task_data json.loads(raw_data) process_task(task_data)注意这里brpop的参数从单个 key 变成了列表[HIGH_QUEUE, NORMAL_QUEUE]Redis 会按列表顺序依次阻塞监听。高优先级队列有任务时立即弹出高优先级为空时等待 5 秒超时然后检查普通队列。这个方案的延迟取决于超时值如果高优队列空且普通队列有任务最长需要等 5 秒。想减少延迟可以设成 1 秒代价是每秒多一次空轮询。生产端则只需按任务类型推入不同队列def publish_task_with_priority(task_type: str, payload: dict, priority: str normal): task { task_id: f{task_type}-{int(time.time() * 1000)}, type: task_type, payload: payload, priority: priority, created_at: time.time() } queue_key HIGH_QUEUE if priority high else NORMAL_QUEUE r.lpush(queue_key, json.dumps(task, clsTaskEncoder, ensure_asciiFalse)) return task[task_id]这个方案的缺陷是高优任务可能被持续生产导致普通队列饥饿。实际应用中要么对高优队列做速率限制要么在高优任务处理完一定数量后主动切回普通队列保证两边都有进展。你在 Demo 基础上做二次开发时这两条路可以都试试。4. 消费端可靠性ack、重试与幂等设计4.1 消费即删除的隐患任务处理失败后为什么找不回来List 方案最大的痛点是BRPOP弹出任务的同时就把它从 Redis 里删除了。如果process_task中途崩溃或抛异常这个任务就永远丢失了。Demo 代码里虽然用try-except捕获了异常并打日志但任务本身没有重新入队这对于「丢任务不可接受」的场景是致命的。常见补救方案是「备份队列 重新入队」。在process_task抛出无法恢复的异常时把原始任务数据推入一个task_queue_dead队列方便事后人工处理或者写定时脚本重放def run_worker_safe(): while True: result r.brpop(QUEUE_KEY, timeout5) if result is None: continue _, raw_data result try: task_data json.loads(raw_data) process_task(task_data) except Exception as e: # 处理失败推入死信队列保留原始数据 r.lpush(task_queue_dead, raw_data) print(ftask moved to dead queue: {e}) def replay_dead_tasks(max_count: int 100): # 从死信队列恢复任务重新投入主队列 for _ in range(max_count): raw_data r.rpop(task_queue_dead) if raw_data is None: break r.lpush(QUEUE_KEY, raw_data) print(freplayed up to {max_count} dead tasks)死信队列的 key 命名建议带日期比如task_queue_dead_20250617这样后续排查问题时可以按时间段定位。重放逻辑里加max_count是为了防止死信队列里积累了大量坏数据时一次性全部重放导致主队列被污染。4.2 延迟重试不要立即重试同一个任务用退避策略处理任务失败的另一个常见诉求是「自动重试几次」。但立即重试通常没有意义——如果任务是因为依赖的外部接口暂时不可用而失败立刻重试大概率还是失败。常见的做法是记录重试次数判断次数是否超限未超限则推回队列尾部而非头部。def process_with_retry(task_data: dict, max_retries: int 3): try: process_task(task_data) return True except Exception as e: retry_count task_data.get(retry_count, 0) if retry_count max_retries: task_data[retry_count] retry_count 1 # 使用 rpush 推入右端队列是左进右出右端是尾部 r.rpush(QUEUE_KEY, json.dumps(task_data)) # 记录下次执行时间到 ZSet用于延迟控制 next_run_ts time.time() (2 ** retry_count) r.zadd(task_retry_schedule, {json.dumps(task_data): next_run_ts}) print(ftask scheduled for retry #{retry_count 1}) else: r.lpush(task_queue_dead, json.dumps(task_data)) print(ftask exceeded max retries, moved to dead queue) return False结合 ZSet 的task_retry_schedule可以让任务在指定时间后才被重新消费一个独立的调度线程每隔几秒扫描 ZSet 到期任务再推入主队列。这个方案比简单的推回队列尾部更精细控制了重试的节奏。指数退避的公式2 ** retry_count含义是第一次重试等 2 秒第二次等 4 秒第三次等 8 秒。你可以按业务需求调整基数比如改成10 ** retry_count或2 ** (retry_count 1)。4.3 幂等设计消费者重复执行任务时如何保证数据不错即使有重试和死信队列也无法避免「消费端处理成功但返回给 Redis 前连接断开」这种极端情况。此时任务还在 Redis 中会被再次弹出执行所以任务的幂等性要从业务层面保证。最简单的幂等策略是给每类任务生成一个唯一的biz_key例如图片处理任务可以用image_path resolution拼接作为 key在执行前先检查处理结果标记def process_task_idempotent(task_data: dict): biz_key f{task_data[type]}:{task_data[payload].get(image_path)} # 假设结果标记存 Redis处理成功写入标记 if r.exists(ftask_done:{biz_key}): print(ftask already done, skip: {biz_key}) return process_task(task_data) # 处理成功后写入标记带过期时间防止无限膨胀 r.setex(ftask_done:{biz_key}, 3600 * 24 * 7, 1)setex的过期时间根据业务保留时长决定图片处理这种一次性任务保留 7 天足够。如果任务处理结果需要持久保存到数据库幂等应该依赖数据库的唯一约束而不是 Redis 标记——Redis 标记只适合做短时间内的防重。判断幂等标记后直接 return 而不是抛异常这点很重要。否则重复执行的任务会进入重试流程最终堆积到死信队列里干扰真正失败任务的排查。5. 避坑任务队列五个高频翻车现场与排查手册5.1 阻塞时间过长导致 Redis 连接被服务端断开现象消费者跑一段时间后日志里突然全是对应的 TIMEOUT 错误之后再也不消费任务了。原因BRPOP阻塞时间默认是 0表示永久阻塞。Redis 服务端timeout配置如果设了非 0 值空闲连接会被服务端主动关闭。而很多 redis-py 版本在连接被关闭后不会立刻感知要等到下次真正读写时才抛出 ConnectionError。消费者如果一直阻塞在brpop上连接已被服务端断开但客户端毫不知情看起来就是「任务发布了但消费者不消费」。解决连接池加health_check_interval30同时BRPOP的 timeout 不要用 0建议设 30~60 秒。这样即使连接被服务端回收客户端也能在 30 秒内主动发现并重建连接。我在排查这个问题时用redis-cli CLIENT LIST看到消费者的连接idle1200秒立刻意识到是服务端断开后客户端没感知。5.2 多消费者实例导致任务重复消费现象开了两个 Worker 进程同一个任务被打印了两遍处理日志。原因多个 Worker 用BRPOP监听同一个 key 时Redis 保证每个元素只被一个 Worker 弹出但如果 Worker 进程内开了多线程且每个线程各自执行brpop这本身没问题。真正的问题出在「处理成功但 ack 前任务被重新入队」的重试机制上。如果你的消费者在进程内异常退出退出前从队列弹出的任务丢失重启后又从队列读到另一个任务配合外部重试机制就可能重复处理同一逻辑。解决先确认是「同一个任务被消费两次」还是「不同的任务但业务上等价」。前者需要检查是否多个 Worker 连了不同的 Redis 库比如db0和db1后者需要幂等标记兜底。我调试时发现两个 Worker 脚本一个连本机 6379另一个通过配置文件连到了测试环境 Redis两个环境数据同步延迟导致看起来像是重复消费。5.3 序列化时报错Object of type datetime is not JSON serializable现象生产者发布任务时直接抛 TypeError任务根本没有进入队列。原因任务 payload 里包含了datetime对象而标准库json.dumps不支持。解决按 3.1 节的方案自定义TaskEncoder。如果是在消费端反序列化时报错通常是生产端和消费端的编码类不一致生产端用自定义 Encoder 序列化出 ISO 格式字符串消费端json.loads拿到的是字符串而不是 datetime 对象需要手动datetime.fromisoformat转换。建议在任务结构里把时间字段统一存成时间戳浮点数彻底绕开这个问题。5.4 任务积压严重但消费者 CPU 占用极低现象LLEN task_queue显示任务数量持续增长但消费者进程 CPU 只有 2%。原因消费者消费速度跟不上生产速度但因为是串行处理每个任务处理需要 0.5 秒生产端每秒推 100 个任务队列自然越积越长。CPU 低是因为任务大多在等待或 IO 阻塞不是计算密集任务。解决先看process_task里有没有不必要的time.sleep或串行等待。图片处理这种 IO 密集操作可以用ThreadPoolExecutor把工作线程数调大或者用multiprocessing起多个 Worker 进程。消费端改造示例from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers10) def run_worker_concurrent(): while True: result r.brpop(QUEUE_KEY, timeout5) if result is None: continue _, raw_data result executor.submit(handle_task_safe, raw_data) def handle_task_safe(raw_data: str): try: task_data json.loads(raw_data) process_task(task_data) except Exception as e: print(ftask failed: {e})注意线程池方式下任务并发执行时会同时并发访问 Redis 或外部接口连接池的max_connections必须大于线程数否则连接不够时线程会排队效果又退回串行。5.5 Worker 进程崩溃后任务丢失重启恢复困难现象Worker 被手动 kill 或系统崩溃重启后队列里的任务还在但崩溃前正在处理的那些任务消失了。原因BRPOP弹出即删Worker 崩溃时正在执行的任务没有机会入队或标记。解决最彻底的方案是改用 Redis Stream 的XREADGROUPXACK机制。Stream 支持消费者组消息被读取后不会立即删除而是进入 PEL待确认列表只有显式XACK后才会被标记为已完成。Worker 重启后还能从 PEL 里找出未确认的消息重新处理。这个改进值得在 Demo 基础上做一次完整重构def stream_worker(): group_name image_workers consumer_name fworker-{os.getpid()} stream_key stream_task_queue try: r.xgroup_create(stream_key, group_name, id0, mkstreamTrue) except redis.ResponseError as e: # 组已存在则忽略 if BUSYGROUP not in str(e): raise while True: messages r.xreadgroup( group_name, consumer_name, {stream_key: }, count10, block5000 ) for stream_name, entries in messages: for msg_id, fields in entries: try: task_data json.loads(fields.get(data, {})) process_task(task_data) r.xack(stream_key, group_name, msg_id) print(facked message: {msg_id}) except Exception as e: # 不 ACK 的消息会留在 PEL 中可稍后重试 print(fprocess error, message stays in PEL: {e})xreadgroup里的block5000表示阻塞 5 秒count10是每次最多取 10 条消息避免一次取太多导致单条消息处理时间过长影响其他消费者。Stream 方案的代价是需要理解消费者组、消息 ID、ACK 这几个新概念但换取的是崩溃后不丢消息的可靠性做生产级队列时这笔账是划算的。6. 积压监控与消息确认让队列状态一目了然队列跑起来之后真正决定体验的是你能否随时回答三个问题现在积压了多少任务最快的一条任务等了多久有没有任务死在半路没人管这个小节我分享一个非常实用的监控小脚本能让你在十分钟内把这些信息全部显示出来。监控的核心是LLEN和LRANGE的组合。LLEN告诉你积压数量但数量本身意义有限——如果积压 1000 条但最早的任务只是 5 秒前进来的说明消费速度勉强跟得上如果最早任务是 3 小时前进来的那就要告警了。所以脚本里需要扫描队首几条任务的时间戳import json import time import redis r redis.Redis( host127.0.0.1, port6379, db0, decode_responsesTrue, socket_connect_timeout3, socket_timeout3 ) def check_queue_health(): queue_key task_queue dead_key task_queue_dead now time.time() queue_len r.llen(queue_key) dead_len r.llen(dead_key) # 取队尾的 10 条任务队尾是最先被消费的也是等待最久的 oldest_tasks r.lrange(queue_key, -10, -1) max_age 0.0 for raw in oldest_tasks: try: task json.loads(raw) created_at task.get(created_at, now) age now - created_at max_age max(max_age, age) except (json.JSONDecodeError, TypeError): continue print(fqueue_len{queue_len}, dead_len{dead_len}, oldest_wait{max_age:.1f}s) if max_age 300: print(警告: 存在等待超过 5 分钟的任务检查消费者是否存活或阻塞) if dead_len queue_len * 0.1 and dead_len 20: print(警告: 死信队列占比过高需要人工干预) def replay_stuck_tasks(max_age_seconds: int 300): # 将等待时间超过阈值的任务重新投入高优队列 queue_key task_queue high_queue task_queue_high raw_data r.rpop(queue_key) if raw_data: r.lpush(high_queue, raw_data) print(fmoved old task to high priority queue: {raw_data[:80]}) if __name__ __main__: check_queue_health()扫描范围-10, -1是 Redis List 切片语法表示从倒数第 10 个元素到最后一个元素也就是队列中最先被生产、最有可能积压最久的那部分任务。这里不需要扫描全部取尾部样本估算最大等待时间就够了量大时全量扫描会拖垮 Redis 实例。有了这个脚本你可以把它丢进 crontab 每分钟跑一次积压和延迟告警就有了最基础的依据。更进一步的做法是给死信队列加上一个独立的统计追踪脚本记录每条死信任务的原始类型和失败次数这样每天上班先看眼统计文件就知道昨天夜里系统踩了什么坑。实际在使用这份 Demo 时我有一段血泪经验最初我的监控脚本只统计队列长度觉得只要长度在涨就马上加消费者结果加了三个消费者之后队列反而越积越长最后才发现单个任务内部还在调用另一个 Redis 队列等待回调形成了嵌套阻塞间接把消费线程全卡死了。从那以后我每次改队列逻辑都会强制走一遍全链路压测用redis-cli MONITOR观察实际的命令耗时而不是盯着队列长度做判断。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
