如何在 Locust 中识别验证码而不影响统计数据

在 Locust 里做验证码识别，有两个地方必须避开：统计数据，以及每次迭代都会走到的代码路径。Locust 会记录每一个通过 self.client 发出的请求，所以用它来识别验证码，就会把识别工具的响应时间混进你想读的那份报告里。而写在 task 内部的识别调用，会在每个用户的每一次迭代中都执行一遍。只要清楚边界在哪里，这两件事都不难避免。

你需要什么

• Locust 2.x，运行在 Python 3.10 或更高版本上，这也是 CapSkip 客户端的要求。
• CapSkip 运行在一台 Windows 机器上，Python 客户端和你的 locustfile 装在一起。
• 受保护表单的 sitekey 和页面 URL，由外部传入，而不是让每个用户各自重新获取。
• 如果 Locust 运行在识别工具本机之外的任何地方，就需要 Server 模式，容器和每一台 worker 机器都算在内。这只是连接设置里的一个选项。

# pip install capskip
pip install -U locust capskip

第 1 步：不要让识别请求走 self.client

这一点会悄无声息地毁掉一份报告。Locust 的文档明确说明了 self.client 是什么：一个 HttpSession 实例，它是 requests.Session 的子类和封装，额外做的事情就是把请求结果上报给 Locust。成功与失败、响应时间、响应长度、名称。凡是经它发出的请求，都会出现在统计表里。

一次验证码识别要花几秒。你的应用接口只要几毫秒。把两者放进同一张表，你的 95 分位数就变成了验证码耗时的度量，而这并不是任何人想要的数字。

好消息是，解决办法就是什么都不用做。CapSkip 客户端有自己的 HTTP 传输层，从不碰 self.client，所以识别过程默认对 Locust 的统计不可见。Locust 只记录走它自己 session 的请求。

# pip install capskip
import os
from capskip import CapSkip

# CAPSKIP_HOST is the solver machine. 127.0.0.1 only works when
# Locust and CapSkip run on the same Windows box.
solver = CapSkip(
    host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.environ.get("CAPSKIP_PORT", "8080")),
    apiKey=os.environ.get("CAPSKIP_API_KEY", "capskip"),
)

def fresh_token(sitekey, page_url):
    result = solver.recaptcha(sitekey=sitekey, url=page_url)
    return result["code"]   # the token, and Locust never sees it

如果你更想直接对着原始接口自己写，规则也一样：用普通的 requests.Session，而不是 self.client。直接用 requests 库发出的请求不会被 Locust 记录，这正是这里想要的效果。这两个接口的说明见 CapSkip API 文档.

唯一要反着做的情况，是你有意在对识别工具本身做压力测试。这时就让请求走 self.client 并带上 name 参数，这样所有识别会归到同一行，而不是每个 sitekey 一行，然后把那一行和其余数据分开来读。

第 2 步：你的识别调用到底会跑多少次？

Locust 给了你四个可以放置它的位置，彼此之间相差几个数量级。先算清楚倍数，再决定放哪里。

对绝大多数人来说，默认答案是 on_start。用户开始运行时会调用 on_start，于是每个模拟用户拿到自己的 token，在自己的会话里持有它，也不需要在 greenlet 之间传递任何东西。这也最接近真实用户的行为。

from locust import HttpUser, task, between

class Signup(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        # One solve per simulated user, off the statistics.
        self.token = fresh_token(SITEKEY, PAGE_URL)

    @task
    def submit(self):
        self.client.post("/signup", data={
            "email": "[email protected]",
            "g-recaptcha-response": self.token,
        })

五百次识别听起来很贵，因为在按次计费的服务上确实如此。这也是大多数压测指南削足适履地只识别一次再共享结果的原因。而当识别工具跑在你自己的硬件上，这个数字就不再是预算问题，而是容量问题，后者要好回答得多：把并发压上去，看着这台机器就行。

第 3 步：test_start 在每个节点上都会触发，而不是每次运行只触发一次

这是 Locust 里最容易让人意外的细节，表现出来就是识别次数正好是某个数的整数倍。文档说，新的压测启动时，test_start 会在每个节点上触发。用四个 worker 进程运行，你就有五个节点，所以放在普通 test_start 监听器里的识别会执行五次。

通过检查 runner 类型来加上判断。Locust 正是为此提供了 MasterRunner 和 WorkerRunner，它自己的分布式指南里给出的做法就是判断当前处在哪一种上。

from locust import events
from locust.runners import WorkerRunner

@events.init.add_listener
def on_init(environment, **kwargs):
    # Workers listen. Registering here runs before the test starts.
    if isinstance(environment.runner, WorkerRunner):
        environment.runner.register_message("captcha_token", take_token)

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    # The master solves once and broadcasts. Workers skip this.
    if not isinstance(environment.runner, WorkerRunner):
        environment.runner.send_message(
            "captcha_token", fresh_token(SITEKEY, PAGE_URL)
        )

def take_token(environment, msg, **kwargs):
    environment.shared_token = msg.data

关于这段代码有两点。处理函数的签名是固定的：它接收 environment、message 和关键字参数，负载通过 msg.data 传来。另外，如果某个处理函数会执行较长时间，注册时把 concurrent 设为 True，让它在自己的 greenlet 里运行，而不是阻塞 Locust 的心跳和其他系统消息。只存一个字符串的处理函数不需要这样做，在内部执行识别的处理函数则需要。

在动手实现之前先读下一节，因为每次运行只识别一次通常本来就是个错误的目标。

第 4 步：token 撑不过一场长时间的压测

一个 reCAPTCHA token 的有效期大约两分钟。而一次压测通常不止两分钟。于是那套看起来很整洁的架构，也就是在测试开始时识别一次再广播给所有 worker，会造出这样一次运行：前几分钟一切正常，之后全部因为 token 过期而失败，而这些失败还会被算到你的应用头上。

这种失败值得一眼就认出来，串联队列任务的人也常栽在同一个问题上。完整的分析见这篇指南： reCAPTCHA token 过期.

所以，只有在测试时间很短，或者 token 只在爬坡阶段用一次而不是每次迭代都用的时候，才采用共享 token 的做法。其他情况就在 on_start 里按用户各自识别；如果是跑好几个小时的稳定性测试，就在用户内部定时刷新。

import time

class Signup(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        self.token = fresh_token(SITEKEY, PAGE_URL)
        self.solved_at = time.monotonic()

    @task
    def submit(self):
        # Refresh before the token ages out, not after it fails.
        if time.monotonic() - self.solved_at > 90:
            self.token = fresh_token(SITEKEY, PAGE_URL)
            self.solved_at = time.monotonic()
        self.client.post("/signup", data={
            "g-recaptcha-response": self.token,
        })

用九十秒而不是一百二十秒，这样刷新发生时 token 还没失效。

第 5 步：Locust 用的是 gevent，所以要用同步客户端

Locust 把每个用户放在各自的 greenlet 里运行，基于事件，底层用的是 gevent。它的文档特别指出，正是这一点让你可以把测试写成普通的阻塞式 Python 代码，而不必使用回调。阻塞在这里是原生风格，所以该用的是普通的 CapSkip 客户端。

不要在 locustfile 里使用 AsyncCapSkip。在 Python 中它是真正的异步实现，而不是一个别名，因此在 asyncio 程序里它才是合适的客户端，而 Locust 并不是 asyncio 程序。这里没有事件循环在等它，而在每个用户的 greenlet 内部各起一个事件循环，等于绕一大圈才换回 gevent 本来就已经做到的事情。

轮询行为在这里也有帮助。客户端并不是按固定间隔轮询的。它从四分之一秒开始，逐步退避到 pollingInterval，所以识别得快就返回得快，而不是白等一个固定延迟。当五百个 greenlet 一起爬坡时，这点差别几乎就是你整个爬坡阶段的差别。如果你在别处的 asyncio 程序里确实需要一次识别一批验证码，那种情况可以参考 在 Python 中并行识别验证码.

第 6 步：把识别工具放在 worker 能访问到的地方

压力发生器很少和别的东西挤在同一台机器上。它们会独占一台机器，或者几台，或者一组容器，正是为了让压力足够真实。CapSkip 运行在 Windows 上，而你的 worker 未必如此。

连接模式有两种。Local 模式绑定 127.0.0.1，只响应本机。Server 模式绑定你的内网地址或公网 IP，这样另一台机器、一台容器宿主机或者一个托管 runner 就能通过 API 访问同一台 Windows 机器。两者都位于 连接设置，而 Server 模式只改变识别程序监听在哪个地址上。硬件还是你自己的，识别也依然不计量。

地址从环境变量里读，而不要写在 locustfile 里。Python 客户端不会自己读取 CAPSKIP_HOST、CAPSKIP_PORT 或 CAPSKIP_API_KEY，所以上面创建 solver 的代码会读取它们并传进去，这样同一个文件不用改动，在你的笔记本上和在一整组 worker 上都能用。

常见错误及其含义

最后这一条有专门的指南，因为同一个响应既覆盖密钥缺失的情况，也覆盖密钥填错的情况： 如何修复 ERROR_WRONG_USER_KEY.

常见问题

简短版结论

用 CapSkip 客户端来识别，这样请求既不会经过 self.client，也不会进入你的统计数据。把调用放在 on_start 里，让每个模拟用户拿到一个 token；如果单次运行超过大约九十秒，就在 task 内部刷新它。如果你确实要在 test_start 监听器里识别，记得加上 WorkerRunner 判断，因为那个事件在每个节点上都会触发。坚持使用同步客户端，因为 Locust 的并发靠的是 gevent。只要压力发生器不在识别工具所在的那台机器上，就把 CapSkip 切到 Server 模式运行。

• 客户端的完整接口以及它覆盖的每一种验证码类型，请见 Python 验证码识别页面.
• 勾选框挑战本身的说明，请见 reCAPTCHA v2 识别页面.

在决定爬坡规模之前，还有一点需要先确定：CapSkip 作为一款 无限量验证码识别工具 运行在你已经拥有的硬件上，所以一千个模拟用户各自识别自己的挑战，花费和十个用户完全一样。