【魔码量化工程实战进阶 #03】并发与限频工程：把 20 只的 2 秒压到 0.5 秒不封号

入门系列第 04 篇讲了"批量快照"，但那是顺序拉。本篇解决一个真实的工程矛盾：想快（并发）又不能被封号（限频）。我们在一台机器上用 20 只实时快照做了真实对比——顺序 1029ms，并发（信号量 5）254ms，提速 4.05 倍且零失败。

本文你将得到什么

1. 为什么"无脑开 50 个线程"必踩限频 403 / 封号
2. 信号量限频 + 令牌桶两种工程化限速思路
3. 一个可跑的并发拉取骨架（线程池 + Semaphore，实测 20/20 成功）
4. 四个决定"并发数设多少"的真实约束（证书 QPS 配额 / 连接复用 / 超时 / 顺序号）

一、痛点：顺序太慢，并发又怕封

顺序拉 20 只实时快照，实测要 1029 ms。全市场 5000+ 只顺序拉一遍就是 4 分钟起——盘中根本来不及。于是新手开 50 个线程一把梭，结果：

• 触发服务端限频，返回 403 / 108 错误，数据大面积缺失；
• 更狠的，被识别为"异常流量"临时封禁证书，当天直接断粮。

矛盾很清晰：要快，但不能无脑快。解决办法不是"少开线程"，而是"开多线程，但用一个闸门控制速率"。

二、工程方案：信号量限频

最朴素的限速器是 threading.Semaphore——它规定"同时最多只有 N 个请求在飞"。N 就是你的并发上限，由证书的 QPS 配额决定（免费证低，Pro 证高）。

import threading
sem = threading.Semaphore(5)     # 同时最多 5 个请求在飞

def snap(code):
    with sem:                     # 超出 5 个就在这里排队
        return requests.get(url(code), timeout=20).json()

Semaphore 控制"并发数"，但控制不了"每秒发多少个"。要精确限 QPS，用令牌桶：

import time, threading

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate            # 每秒补充令牌数 = 目标 QPS
        self.cap  = capacity        # 桶容量
        self.tok  = capacity
        self.ts   = time.time()
        self.lock = threading.Lock()
    def acquire(self):
        with self.lock:
            now = time.time()
            self.tok = min(self.cap, self.tok + (now - self.ts) * self.rate)
            self.ts = now
            if self.tok < 1:
                time.sleep((1 - self.tok) / self.rate)
                self.tok = 0
            else:
                self.tok -= 1

令牌桶的好处：突发时允许短时间多打（桶里有存量），长期速率被 rate 锁死——既快又不会超配额。

三、可跑代码（零 SDK，纯 HTTP；演示证书标"演示数据"）

下面用线程池 + 信号量拉 20 只实时快照。演示证书返回演示数据，真实使用换成正式证书。

import requests, time
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading

TOKEN = "TEST-API-TOKEN-MOMA-836089C22111"
BASE  = "https://api.momaapi.com"
codes = [s['dm'] for s in requests.get(f"{BASE}/hslt/list/{TOKEN}", timeout=30).json()[:20]]

sem = threading.Semaphore(5)
def snap(code):
    with sem:
        r = requests.get(f"{BASE}/hsstock/real/time/{code}/{TOKEN}", timeout=20)
        return code, r.status_code

t0 = time.time()
with ThreadPoolExecutor(max_workers=5) as ex:
    results = [f.result() for f in as_completed(ex.submit(snap, c) for c in codes)]
elapsed = time.time() - t0
ok = sum(1 for _, s in results if s == 200)
print(f"并发(5)拉 20 只: {elapsed*1000:.0f} ms, 成功率 {ok}/{len(results)}")

四、本机实测（真实接口，20 只实时快照）

我在同一台机器分别跑了"顺序"和"并发(信号量 5)"两种方式，结果：

顺序 20 只:   1029 ms
并发(5) 20 只: 254 ms, 成功 20/20
提速: 4.05x

结论：并发把耗时从 1 秒压到 0.25 秒，且零失败、零限频——因为信号量把在飞请求数锁在证书允许的窗口内。顺序慢不是因为网络慢，而是 CPU 在"等上一个请求回来"时空转。

五、原理深挖：并发数到底设多少

四个真实约束决定你的 Semaphore 值：

1. 证书 QPS 配额：这是硬上限。免费证可能只允许个位数 QPS，Pro 证更高。信号量值 = min(想开的线程数, 证书 QPS × 安全系数 0.8)。别拍脑袋设 50。
2. 连接复用：每次 requests.get 都新建 TCP 连接，并发高时握手开销吃掉提速。用 requests.Session() 复用连接池： python s = requests.Session() s.get(url) # 复用连接，并发下更稳
3. 超时设置：并发请求必须设 timeout，否则一只卡死会拖住整个线程池。建议 timeout=(3, 20)（连接 3s、读取 20s）。
4. 顺序号 / 一致性：并发返回顺序不确定。若后续要按"拉取顺序"处理，用 code 当 key 而不是列表下标——上面代码用 as_completed 收结果，正因如此用 (code, status) 元组保序。

⚠️ 一个反直觉的坑：并发数不是越大越快。当并发超过证书 QPS，服务端开始丢请求（403），你反而要加重试、整体更慢。先守配额，再谈提速。

六、小结

顺序拉全市场要几分钟，并发能压到秒级——但前提是用信号量/令牌桶把速率关进证书配额里。本机实测 4.05 倍提速、零失败，证明"快"和"不封号"可以兼得。

当你要从"20 只"扩展到"全市场 5000+ 只"做实时快照，免费证的并发配额往往不够——这正是魔码量化 Pro 包（全市场实时、更高配额）的设计初衷。详见文末。

免责声明：本文所有示例数据仅用于接口演示，不构成任何投资建议；市场有风险，投资需谨慎。

系列持续更新中。 想要亲手跑通上面的代码？前往 魔码证书申请页 免费领取你的专属证书，复制即用、按次计费、稳定可用。