把一个大 JSON 塞进 Redis 之后，很多团队会不自觉地陷入一个陷阱：既然数据已经在 Redis 里了，那就顺手在同一个请求链路里把它处理掉吧。结果接口 RT 从几十毫秒飙到几秒，Redis 网络带宽被大 Value 吃满，用户端转圈转到超时。

大 JSON 本身就已经是 Redis 的性能杀手。一个 200KB 的 JSON 写入，上行网络延迟就能干到几十毫秒；如果是 300KB 的 BigKey，GET 一次的解压、反序列化、传输，RT 根本不可能稳定在 1ms 级别。如果再在这个链路里叠加耗时任务，等于把网络瓶颈和计算瓶颈串联放大。

大 JSON 只负责“存”，耗时任务通过 Redis Stream 异步“解”

。API 把大 JSON 的引用（或 ID）写入 Stream 后立即返回，后台 Worker 按自己的节奏消费并处理。本文从实操角度拆解这套方案的关键设计。

一、先认清一个前提：大 JSON 本身不该被塞进 Stream 消息体

Redis Stream 的消息体同样是存在 Redis 内存里的。如果把 200KB 的 JSON 直接作为 XADD 的字段值，消费者每次 XREADGROUP 都会把这 200KB 拉回本地网络，PEL（待处理条目列表）里积压的每一条消息都是一颗带宽炸弹。

这样 Stream 的 PEL 内存占用可控，消费者组的吞吐不会被大 Value 拖垮，同时保留了 Redis Stream 的崩溃恢复能力。Redis 官方教程中的 Job Queue 示例也是这个思路：Stream 消息里只放 

 存在独立的 Key 中，Worker 消费时再 

二、Stream 解耦的核心三件套：XADD、XREADGROUP、XACK

def submit_job(payload: dict):
    job_id = str(uuid4())
    # 大 JSON 独立存储，不进入 Stream
    redis.json().set(f"job:payload:{job_id}", "$", payload)
    # Stream 只传引用
    redis.xadd("jobs:pending", {"jobId": job_id, "type": "heavy_parse"})
    return {"job_id": job_id, "status": "queued"}

 返回的 Stream ID 可以作为任务的投递凭证。API 在几毫秒内完成写入并返回，用户端拿到 

 后通过轮询或 WebSocket 获取结果。

消费者侧：消费者组保证“每个任务只给一个 Worker”

 就够了，但只要部署超过一个实例 hnying.com，就必须用

 从同一个组读取，Redis 保证同一条消息不会被组内两个消费者同时拿到。

try:
    redis.xgroup_create("jobs:pending", "workers", id="0", mkstream=True)
except ResponseError:
    pass  # 组已存在

while True:
    entries = redis.xreadgroup(
        "workers", "worker-1",
        {"jobs:pending": ">"},
        count=1, block=5000
    )
    for stream_name, messages in entries:
        for msg_id, data in messages:
            process_heavy_task(data["jobId"])
            redis.xack("jobs:pending", "workers", msg_id)

。没有 XACK 的消息会一直留在 PEL 里，成为“僵尸消息”，既不会被重新投递，又会持续占用内存。

三、崩溃了怎么办？PEL 与故障恢复的真实代价

Stream 比 List 可靠的地方在于：消费者挂了，消息不会丢。但这份可靠不是免费的——

，积压的未确认消息会消耗可观的内存，并拖慢 

，只读新消息。这意味着如果 Worker 在处理某条消息时崩溃，重启后它

再看到那条消息——消息卡在 PEL 里，直到有人主动认领。

# 阶段一：bdmingdian.com
while True:
    pending = redis.xreadgroup("workers", "worker-1",
                               {"jobs:pending": "0",}, count=100)
    if not pending or not pending[0][1]:
        break
    for msg_id, data in pending[0][1]:
        process_and_ack(msg_id, data)

# 阶段二：再读新消息
while True:
    new = redis.xreadgroup("workers", "worker-1",
                           {"jobs:pending": ">"}, block=5000)
    ...

 查询的是当前消费者已经读取但未确认的历史消息。先把这些处理完，再切换到 

其他 Worker 挂了：XAUTOCLAIM 做“工作窃取”

如果某个 Worker 永久下线，它 PEL 里的消息需要被其他实例接管。Redis 6.2+ 提供 

，原子地扫描并认领空闲超过指定时间的消息：

redis.xautoclaim("jobs:pending", "workers", "worker-2",
                 min_idle_time=60000, start="0-0", count=10)

每个 Worker 可以跑一个后台协程，每 30 秒执行一次，自动捡起故障节点遗留的任务。注意 

 要大于你正常任务的最长处理时间，否则会把还在跑的任务误抢过来。

四、大 JSON 场景下的两个额外考量

如果某条消息因为 JSON 格式错误或字段缺失导致 Worker 每次处理都抛异常，它会不断被 XACK 失败、被 XAUTOCLAIM 重新认领、再崩溃，形成无限循环。

 字段可以用来识别这种情况：投递次数超过阈值（比如 3 次）的消息应该被移入

Redis 官方教程中的死信队列做法是：达到最大重试次数后，把 

、错误原因、原始载荷摘要写入一个独立的 

Stream 是 append-only 的，不会自动删除已消费的消息。大 JSON 任务如果持续写入，Stream 本身会无限膨胀。需要用 

redis.xtrim("jobs:pending", maxlen=10000, approximate=True)

 让 Redis 按 Radix Tree 节点批量裁剪，比精确裁剪快得多。对于已经进入死信队列的条目，也要设定保留策略，避免死信 Stream 变成新的内存黑洞。

五、什么情况下不该用这套方案

中等吞吐、需要消费者组协调、已经在用 Redis

的场景。如果你的任务量达到每秒数万条以上 gdhongjian.com，或者需要跨机房持久化、复杂路由、长期消息保留，Stream 不是 Kafka 或 RabbitMQ 的对手。

但绝大多数“大 JSON + 耗时解析/计算”的业务场景，吞吐远没到那个量级。用 Stream 解耦的收益很直接：

别让它的处理也留在 Redis 的同步链路上

。XADD 一行代码的事，剩下的交给 Stream 和 Worker 慢慢消化。

大 JSON 写入 Redis 后，怎样用 Stream 拆分并解耦耗时任务？

数据库

架构设计

后端

大JSON存入Redis后，若在请求链路中直接处理会导致RT飙升和网络带宽瓶颈。正确方案是分离存储与调度：大JSON独立存储，Stream消息仅携带jobId，通过XADD、XREADGROUP、XACK实现异步解耦，配合XAUTOCLAIM处理故障恢复，避免毒药消息，控制内存增长，确保接口毫秒级响应。

网络延迟

WebSocket

路由

Redis

Kafka

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

大 JSON 写入 Redis 后，怎样用 Stream 拆分并解耦耗时任务？

大 JSON 写入 Redis 后，怎样用 Stream 拆分并解耦耗时任务？

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐