批量任务排队不动，通常不是 Wan-Streamer v0.2 本身挂了，而是并发上限和单次超时互相拖累：并发开得大，单条任务更容易撞超时；超时设得短，失败任务进入重试又把队列占住，后面新提交的任务就一直排在队列里不动。比较稳的处理顺序是：先用 5 条任务把并发从 1 往上逐档加压，找到第一个出现排队或失败的档位；再按单条任务的实际耗时分布定超时值，并把超时和重试次数放在一起调；最后给重试加上幂等判断，避免重试把同一条任务生成多份产物。

先不要直接调大并发。用 5 条任务从并发 1 逐档加压，记录每档的成功数、失败数和首次失败的返回原文，找到开始出现排队或失败的第一个并发值；再按单条任务耗时的 p90 或 p99 定超时，超时值和重试次数要成对调整，重试前按任务标识查一次状态，避免同一任务产出多份结果。队列已经积压时，先降并发、拆批次把吞吐恢复起来，再看队列深度和最老任务等待时长判断是否恢复正常。不同机器和任务体量的可用区间差别较大，需要在自己环境里重测。

用 5 条任务从并发 1 开始加压

这一档测试的目的是找到「第一个开始排队或失败的并发值」，不是压出系统上限。任务条数用 5 条即可，任务内容尽量选体量接近日常的最大那批，这样测出来的拐点才有参考价值。下面的脚本只是加压骨架，命令入口、参数名和日志路径请换成自己环境里的实际写法，替换项在注释里标出来了。

# 1) 准备 5 条任务标识，换成自己的任务列表
printf 'task-%s\n' 1 2 3 4 5 > /tmp/ids.txt

# 2) 按并发档位循环，xargs -P 控制并发数
#    wan-streamer submit 换成实际调用方式（CLI 或 HTTP 都一样）
for P in 1 2 3 4 5; do
  cat /tmp/ids.txt | xargs -I{} -P "$P" sh -c '
    start=$(date +%s.%N)
    wan-streamer submit `--input` "{}" > /tmp/out-{}.json 2>&1
    code=$?
    end=$(date +%s.%N)
    echo "P='"$P"' task={} exit=$code cost=$(echo "$end - $start" | bc)"
  ' >> "/tmp/run-p${P}.log" 2>&1
done

每跑完一档，把该档日志里的成功数、失败数、单条耗时和失败返回原文填进同一张表。返回原文不要只写「失败」，要把客户端或日志里那一行原始报错整段贴进来，后面判断是被限流、队列满还是超时全靠它。

判定规则很简单：某一档开始出现失败，或者任务提交后长时间停在排队状态（日志里有提交时间但没有开始执行时间），就把这一档的前一档记为当前可用并发。首次失败的返回原文单独存一份，例如常见的限流、队列已满、请求超时三类，处理方向完全不同，混在一起看会误判成「并发太高」。

统计单条任务的耗时分布再定超时

超时值不要凭感觉填一个整数。先跑一批单条任务，把「提交时间」到「产物落盘时间」的墙钟耗时记下来，一行一个秒数，写入 cost.log；日志打点建议直接记录开始和结束时间戳，而不是依赖中间阶段的日志顺序推算。样本量不用很大，但要覆盖大小任务，几十条通常够看出量级。

# cost.log 每行一个单条任务的耗时（秒）
sort -n cost.log | awk '{a[NR]=$1} END{
  printf "p50=%.2f p90=%.2f p99=%.2f max=%.2f\n",
    a[int(NR*0.5)], a[int(NR*0.9)], a[int(NR*0.99)], a[NR]
}'

取值的保守做法是以 p90 为基线，最大任务体量的场景看 p99；如果 p99 明显高于 p50 数倍，说明存在长尾任务，这时超时阈值要跟着长尾走，或者把长尾任务单独拆批次，而不是把整体超时拉长让所有任务一起等。超时和重试次数要成对给，下面这组是常见的组合思路，具体数值按自己的耗时分布替换：

• p90 约 60 秒左右：超时 180 秒、重试 1 次，留出约两倍余量，重试只兜偶发抖动。
• p90 约 300 秒、p99 接近 900 秒：超时 600 秒、重试 0 到 1 次，优先把长尾任务单独拆出去跑。
• 超时设得比 p50 还小：等于人为制造失败，重试队列会很快堆起来，这种配置要直接排除。

给重试加上幂等判断

排队不动的另一个常见原因，是重试本身把队列填满了。做法是给每次提交算一个稳定的任务标识，重试前先查这个标识对应的状态，已经在排队、执行中或已完成的就不再重复提交。标识建议由任务参数做规范化后哈希得到，参数排序后再算，避免字段顺序不同导致同一任务算出两个标识。

import hashlib, json

def task_key(payload):
    raw = json.dumps(payload, sort_keys=True, ensure_ascii=False)
    return hashlib.sha1(raw.encode()).hexdigest()[:16]

def submit_once(client, payload):
    key = task_key(payload)
    st = client.get_status(key)   # 通用骨架，字段名按实际返回替换
    if st and st.get('state') in ('queued', 'running', 'done'):
        return st                 # 已在队列或已完成，直接复用，不再提交
    return client.submit(payload, idempotency_key=key)

产物侧也要能识别重复：给输出文件统一带上任务标识前缀，例如 out/<key>__<seq>.json，事后按前缀计数就能看出哪些任务出了多份产物。检查方式可以是扫目录统计前缀重复次数，也可以按产物内的任务标识字段去重。这一步放在每次批量跑完之后做，比事后翻日志快得多。

ls out/ | sed 's/__.*//' | sort | uniq -c | awk '$1 > 1'

队列积压时的降级方案

队列已经堆起来时，先不改代码，用两种配置层面的降级恢复吞吐：一是把并发降回上一档已验证可用的值，二是把一次提交的大批次拆成多个小批次串行提交，中间留出间隔。两种方式都是为了减少同一时刻在跑的任务数和重试触发次数，属于止血，不是性能优化，积压清空后要再回到前面的加压流程确认可用区间。

# 降级示例，参数名按实际入口替换
wan-streamer submit `--batch` ids.txt `--concurrency` 2 `--timeout` 180
# 拆批次：把 ids.txt 每 10 行切一个文件，依次提交
split -l 10 ids.txt /tmp/chunk-
for f in /tmp/chunk-*; do
  wan-streamer submit `--batch` "$f" `--concurrency` 2 `--timeout` 180
  sleep 5
done

降级后要盯住几个指标：队列深度是否持续下降、最老任务的等待时长是否在缩短、失败类型里超时类占比是否回落、新提交的任务能否在单条 p90 时间内开始执行。恢复正常的判定建议是：连续观察几个窗口，队列深度单调下降到接近 0、新任务从提交到开始执行的等待时长回到 p90 量级、日志里不再出现新的超时或队列满返回。三项都满足再考虑往上加并发，一次只动一档。