1. 需求拆解与方案选型1.1 这个需求到底在解决什么问题批量登录1000个用户并获取token这听起来像是接口测试、性能测试或者数据初始化场景里的脏活累活。我之前在帮团队搭自动化测试数据平台时就遇到过一模一样的诉求线上有1000个测试账号需要在短时间内拿到每个账号的登录token塞给后续的压测脚本或者接口自动化用例去用。手工登录1000次完全不现实哪怕每次只要10秒也得快3个小时而且中途还容易手抖出错。更深一层看这个需求真正考验的不是“能不能登录”而是三件事第一面对1000个账号时脚本能不能稳定跑完第二登录失败的时候能不能快速定位是哪一批账号出了问题第三生成的tokens.txt格式能不能直接被下游工具消费。很多人一开始只写了个for循环加requests跑起来才发现有的账号密码错、有的被限流、有的token字段在嵌套JSON里取不到最后文件倒是生成了里面一半是空行这种脚本交出去是要被骂的。所以这篇文章不是给一段能跑的代码就完事而是把完整的方案设计、踩坑点、工程化落地都讲清楚。适合谁看后端开发、测试开发、运维工程师以及所有需要在短时间内准备大量认证态数据的从业者。不管你是想给压测准备token池还是给接口自动化框架造数据这套思路都能直接用。1.2 主流实现方案对比与取舍拿1000个用户名密码去换token技术上有三条路可以走。第一条路是最朴素的接口直连。用Python的requests库或者Java的HttpClient模拟登录接口的请求参数解析响应里的token字段。优点是快、轻、可控性强1000个请求并发起来几秒钟就能发完缺点是要求你清楚知道登录接口的完整调用链比如是否需要先获取验证码、是否需要携带签名参数、是否需要先调一次前置接口拿sessionId。第二条路是UI自动化典型的工具是Selenium和Appium。说实话用一个浏览器自动化去登录1000个账号是灾难级的选择你要管理等页面加载、处理弹窗、应对验证码拖动跑一个账号可能就要几十秒而且浏览器开多了内存直接爆掉。UI自动化的正确使用场景是验证“登录按钮能不能点”“验证码组件是否正常”这类端到端功能测试而不是用来批量造数据。第三条路是混合方案主流程走接口遇到需要验证码的环节再调用打码平台或者自动化识别。这种方案我一般不建议普通团队搞除非你确实没有接口文档只能从UI逆向出请求参数否则维护成本太高。最终我采用的方案非常明确Python requests 并发池配合完善的日志和重试机制。选这个方案有几个硬理由一是requests库足够成熟Session对象能自动管理cookie和连接池二是Python写这类脚本的效率高试错成本低三是后面接Jenkins、接测试框架都非常方便。如果你的技术栈是Java思路完全一样用HttpClient或者OkHttp也能实现同样效果。2. 前置准备与环境搭建2.1 账号数据准备在做批量登录之前第一步永远是整理用户数据。别小看这一步我见过太多人代码写得飞快结果卡在数据上Excel里既有明文密码又有加密密码用户名有前缀空格还有一堆已注销的账号混在里面跑起来全是登录失败排查半天才发现是数据源脏。我建议准备一个CSV或者Excel文件至少包含三列username、password、可选的其他依赖字段。如果登录接口需要手机验证码你还得再准备一个phone列或者expected_captcha列。另一个容易被忽视的点是编码Windows下导出的CSV经常是GBK编码Python默认按UTF-8读会直接乱码。稳妥的做法是统一转成UTF-8 with BOM或者读取时指定encoding参数。数据文件准备好后脚本里要加一个前置校验逻辑读取每一行判断用户名是否为空、密码是否为空、是否有重复账号。重复账号这个问题看起来低级但真实数据里特别容易发生尤其当数据是从多个环境导出来合并的时候。重复账号不仅会浪费一次请求还会让你在最后核对token数量和用户数量时不相等怎么排查都觉得不踏实。2.2 接口信息确认没有接口文档的批量登录就是瞎猜但现实项目里接口文档经常是缺失的。如果你手里只有网页版的登录页面可以打开浏览器开发者工具切到Network面板勾选Preserve log然后手工登录一次观察登录请求的URL、请求方法、请求头、请求体。这一步要重点拿三个信息一是登录接口的完整路径二是必需请求头比如Content-Type、Authorization前缀规则、X-CSRF-TOKEN三是成功响应里token字段的嵌套路径。token字段的嵌套路径我单独拿出来说是因为这是导致解析失败的高频坑。很多后端接口返回不是简单的{token:xxx}而是{data:{access_token:xxx,expires_in:7200}}甚至包裹三层。你要么让后端同事给一份具体返回示例要么自己在脚本里加一个递归搜索函数按关键字去遍历整个JSON找到包含token的字段。这个递归方法我后面会给出代码。另外要确认登录接口是否有频控限制。有些服务对单IP的并发请求有限制超了会返回429或者直接封IP。确认方法很简单在测试环境连续发20个请求观察响应码和响应时间波动。如果出现429你就必须在脚本里控制并发数并把限流作为一等公民来设计。2.3 开发环境准备环境方面没什么高难度但有几个细节值得说。首先建议使用Python 3.9以上版本主要是为了用更现代的类型注解和标准库特性。其次requests必须安装到启动环境里用虚拟环境创建一个独立的工程目录避免污染全局包。除了requests最好再装一个tenacity它的重试装饰器写起来非常干净省得自己手写循环。最后如果用户数据在Excel里openpyxl也建议装上。我个人习惯的工程目录结构是这样的batch_login/ ├── config.yaml ├── data/ │ ├── users.csv │ └── error_users.csv ├── logs/ │ └── login_20250101_120000.log ├── output/ │ └── tokens.txt ├── requirements.txt └── batch_login.py目录结构清晰的好处是脚本跑挂在任何一台服务器上别人接手时不用猜你的文件放在哪里。config.yaml用来存放账号数据文件路径、并发数、重试次数、日志路径等可变参数而不是把这些参数硬编码在代码里这样后面要接入Jenkins做参数化构建时会非常方便。3. 批量登录与token提取的完整实现3.1 单用户登录函数怎么写这是批量任务的基本单元必须写得防御性足够强。一个合格的登录函数至少要有四个能力发送请求、校验响应状态、解析token、处理异常。不要试图让这个函数干太多事情比如不要让它同时负责写日志和重试那会让代码变得又臭又难测。我常用的写法是这样的import requests import logging logger logging.getLogger(batch_login) def fetch_token(session, endpoint, username, password, timeout10): payload { username: username, password: password, } try: resp session.post(endpoint, jsonpayload, timeouttimeout) resp.raise_for_status() data resp.json() except requests.exceptions.Timeout: logger.error(用户 %s 登录超时, username) return None except requests.exceptions.HTTPError: logger.error(用户 %s 收到HTTP错误 %s响应体%s, username, resp.status_code, resp.text[:200]) return None except ValueError: logger.error(用户 %s 返回内容不是合法JSON, username) return None token extract_token_by_key(data, token) if token is None: logger.warning(用户 %s 响应中未找到token响应体%s, username, json.dumps(data, ensure_asciiFalse, indent2)[:500]) return None return token如果你不知道token字段嵌套在哪一层就写一个递归去搜索这是我实际项目里反复用到的函数import re import json def extract_token_by_key(data, key_hinttoken): if isinstance(data, dict): for k, v in data.items(): if k key_hint or key_hint in k.lower() and isinstance(v, str): return v result extract_token_by_key(v, key_hint) if result: return result elif isinstance(data, list): for item in data: result extract_token_by_key(item, key_hint) if result: return result return None这个递归搜索的价值在于当接口返回结构变化时你的脚本不容易因为少了一层嵌套就直接崩掉最多是警告没找到token让你从日志里去看真实结构。3.2 面对1000个用户并发与重试策略单用户登录函数写好以后最直接的做法就是for循环跑1000遍。我可以告诉你结果如果是内网测试环境且接口响应很快比如单请求50毫秒那么1000个用户串行下来大概50秒勉强能接受。但如果接口响应平均是1秒那1000个用户就要17分钟这很明显不划算。再加上可能有重试时间还会翻倍。更合理的方案是使用线程池把并发数控制在一个跟服务端承受能力匹配的范围内。我通常先设置20个并发如果日志显示没有429再逐步提高到50甚至100但最大不超过200。线程池怎么做用concurrent.futures.ThreadPoolExecutor。下面是一段完整的并发登录伪代码骨架from concurrent.futures import ThreadPoolExecutor, as_completed def batch_login(user_items, max_workers20, max_retries3): session requests.Session() results {} passed {success: [], failed: []} def process_one(item): username, password item for attempt in range(1, max_retries 1): token fetch_token(session, endpoint, username, password) if token: return username, token, logger.warning(用户 %s 第 %d 次尝试失败, username, attempt) return username, , exhausted retries with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {executor.submit(process_one, item): item for item in user_items} for fut in as_completed(futures): username, token, err fut.result() if token: passed[success].append((username, token)) else: passed[failed].append((username, err)) return passed这里有几个很容易踩的坑。第一Session对象能否在线程之间共用单纯发请求是可以的但如果你同时修改Session的headers或者cookie就会出现竞态问题。安全做法是每个线程维护自己的Session但如果你的请求是无状态的共享一个也没有大问题。稳妥起见我在生产代码里倾向于在process_one内部创建独立的Session。第二重试时不要毫无延迟地立即重新请求否则会加重服务端压力。推荐使用指数退避比如第一次重试前等2秒第二次等4秒第三次等8秒。tenacity库可以直接实现这个逻辑但如果不想引入额外依赖手写sleep也完全能接受。还要注意一个细节进程崩溃或者突然断电时你不想让已经获取到的token全部丢失所以最好一边跑一边往结果文件里追加而不是等到全部跑完再一次性写入。这样即使中途挂了你也能从已有文件里拿回一部分token继续处理。3.3 生成干净的tokens.txt文件tokens.txt的格式没有统一标准但我以实际使用场景来设计。如果你是拿token去调其它接口最简单的格式就是每行一个token。但如果你有1000个token对应1000个不同用户名后续你想定位哪个token是哪个用户的就得带上用户标识。我强烈建议在文件里增加一列用户ID或者用户名用分隔符隔开比如user_0001|eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9... user_0002|eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...这样你的下游脚本可以直接按行split(|)来获得用户与token的映射。如果你只需要纯token列表也可以另外生成一个tokens_only.txt。我通常会同时输出两份一份带用户映射一份纯token方便不同场景直接使用。写入文件的时候建议用UTF-8编码并加换行符with open(output/tokens.txt, w, encodingutf-8) as f: for username, token in success_list: f.write(f{username}|{token}\n)这里有一个原子写的技巧先把内容写入临时文件tokens.txt.tmp全部写完后再用os.replace重命名为tokens.txt。为什么这么做因为如果你的脚本在写文件过程中被CtrlC打断磁盘上可能留下一个半截文件下游脚本读的时候直接被截断报错。原子替换可以保证文件要么是旧的完整版本要么是新的完整版本不存在中间状态。同时不要遗漏统计汇总信息。我习惯在控制台输出一行简单的结果摘要比如“成功获取token 956个失败44个成功率95.6%”。如果失败数量超过预设阈值脚本应该以非0状态码退出这样接到Jenkins时就能自动打红报警。3.4 完整代码骨架为了让你能基于这份思路快速开工我给出一段精简但可运行的集成代码包含读取CSV、并发登录、写入结果文件以及失败数据导出。import csv import os import time import json import logging import requests from concurrent.futures import ThreadPoolExecutor, as_completed logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) logger logging.getLogger(batch_login) ENDPOINT os.getenv(LOGIN_ENDPOINT, https://your-api.example.com/auth/login) USER_FILE data/users.csv OUTPUT_FILE output/tokens.txt ERROR_FILE data/error_users.csv MAX_WORKERS 30 MAX_RETRIES 3 def extract_token_by_key(data, key_hinttoken): if isinstance(data, dict): for k, v in data.items(): if k key_hint or (key_hint in k.lower() and isinstance(v, str)): return v result extract_token_by_key(v, key_hint) if result: return result elif isinstance(data, list): for item in data: result extract_token_by_key(item, key_hint) if result: return result return None def fetch_token(username, password): session requests.Session() payload {username: username, password: password} for attempt in range(1, MAX_RETRIES 1): try: resp session.post(ENDPOINT, jsonpayload, timeout10) resp.raise_for_status() data resp.json() token extract_token_by_key(data, token) if token: return username, token logger.warning(用户 %s 响应中未找到token响应体%s, username, json.dumps(data, ensure_asciiFalse)[:300]) except Exception as exc: logger.warning(用户 %s 第 %d 次请求异常%s, username, attempt, exc) time.sleep(min(2 ** attempt, 8)) return username, None def load_users(csv_path): with open(csv_path, newline, encodingutf-8-sig) as f: reader csv.DictReader(f) return [(row[username].strip(), row[password]) for row in reader] def main(): users load_users(USER_FILE) logger.info(共加载 %d 个用户, len(users)) success [] failed [] with ThreadPoolExecutor(max_workersMAX_WORKERS) as executor: futures [executor.submit(fetch_token, u, p) for u, p in users] for future in as_completed(futures): username, token future.result() if token: success.append((username, token)) else: failed.append((username, login_failed)) with open(OUTPUT_FILE, w, encodingutf-8) as f: for username, token in success: f.write(f{username}|{token}\n) with open(ERROR_FILE, w, encodingutf-8, newline) as f: writer csv.writer(f) writer.writerow([username, reason]) for username, reason in failed: writer.writerow([username, reason]) logger.info(批量登录完成成功 %d失败 %d成功率 %.2f%%, len(success), len(failed), len(success) / len(users) * 100 if users else 0) if __name__ __main__: main()这段代码的核心节奏很直观读取数据、并发调用、按汇总结果写文件和错误名单。你可以直接拿去跑通一版再根据你们登录接口的真实返回结构调整payload字段名和token搜索关键字。4. 常见问题与排查技巧实录4.1 登录失败率高排查顺序如果你跑完一批数据发现成功率只有60%第一反应不要怀疑代码先按下面的顺序做判断。第一步看错误用户名单区分是单个账号问题还是群体性问题。如果1000个失败账号都是同一后缀多半是这批测试数据建错了部门或者被统一禁用了。第二步看接口返回的具体状态码。401一般是密码错误或token过期相关认证失败403大多是服务端权限策略拒绝可能是IP白名单不对或者账号被锁定429则明确是指请求过多被限流。第三步看响应体里的错误信息。很多后端会返回一个比较规范的error message比如invalid password、user not exists、“验证码错误”。不要只看HTTP状态码就断定问题有时候业务错误码藏在200响应体里。第四步手工拿一个失败账号跑一次接口调试工具对比你的脚本代码。这一步能快速定位是参数问题还是头部信息问题。我踩过最典型的坑是登录接口需要把某次前置接口拿到的X-CSRF-TOKEN放在请求头里脚本里漏掉之后所有登录请求全部报403。4.2 token失效与续签问题辛苦把tokens.txt生成后你可能会发现token很快失效了。有些系统access_token的过期时间只有30分钟等你把1000个token写完前面的可能已经不能用了。这个时候需要考虑“边获取边使用”的模式也就是不要让token离线存储而是生成后直接灌入下游系统。如果你的场景是给压测准备低峰期数据则建议提前把token的过期时间记录下来。很多登录接口返回时会带expires_in字段你可以把过期时间一并写进tokens.txt比如第三列写上“获取时间过期时间”。这样下游在使用token时可以先判断本地时间是否超过过期时间决定是直接使用还是重新调用脚本刷新。JWT类的token还有一个特殊之处很多系统不提供即时失效机制只要服务端不校验签发时间token在过期前都是有效的。所以当发现token一直有效但权限不对可以从签发方的用户角色、权限变更角度去看不要只盯着token本身。续签问题的通用解法是写一个小工具读取现有token并调用refresh_token接口替换到tokens.txt里避免重新走一遍1000个用户的完整登录。这也是我在工程里最常做的优化之一。4.3 token exchange failed错误实测排查在真实的OAuth2.0授权码流程中很多人会遇到“token exchange failed: token endpoint returned status 403 forbidden”这类报错。这个错误本身在说你用授权码换token的请求被token端点拒绝了HTTP代码403。直接排查以下几个点按出现频次从高到低排。第一授权码一次性失效。授权码通常有效期很短可能只有60秒甚至更短如果你在回调页面停留过久或者脚本执行过慢再用这个code去换token必然失败。解决方案是拿到code后立刻发起token请求不要做任何多余操作。第二redirect_uri不一致。OAuth授权服务要求换取token时提供的redirect_uri必须和发起授权请求时填写的redirect_uri保持一致包括协议、域名、端口和路径都不能差一个字符。很多团队在本地调试时从callback地址复制过来的地址带了尾部斜杠服务端匹配不上就报403。第三client_id和client_secret与授权码不对应。这个问题在键管理混乱的多环境项目里很常见你拿着生产环境的client_secret去刷新测试环境的code服务端校验不通过就会拒绝。第四地域或IP限制。一些第三方OAuth服务会根据client的注册国家或IP来限制token交换如果你的服务环境跟client配置的地区不一致就会直接被拒。这个在排查的时候要注意别把精力全放在代码逻辑上。如果你遇到的是GitLab相关报错“login failed. check api token or gitlab version”那通常是API token格式错误或版本不匹配。先检查GitLab版本与API路径版本是否匹配再看Personal Access Token有没有勾选对应api的scope这是两个最常见的根因。4.4 文件生成、编码与权限坑tokens.txt生成之后出现空文件、乱码或者权限拒绝也是五花八门的坑汇总点。空文件一般有两种原因一种是脚本还没写完就被中断结果文件初始创建后没有任何内容写入另一种是成功列表确实为空也就是所有登录都失败。最后核对一下统计日志就能区分。乱码问题绝大多数是编码不一致。写文件时你用UTF-8但下游工具用GBK打开中文用户名就变成乱码。解决办法是统一约定编码markdown文档里明确写明tokens.txt是UTF-8编码。如果下游工具比较老旧只认GBK那就写入时用GBK但需要在代码里显式改成encodinggbk。权限问题在Linux服务器上特别常见。脚本以root用户生成tokens.txt别的应用用普通用户运行后读取失败。建议把输出目录的所有者改成运行Web服务的用户或者设置目录权限为755、文件权限为644。token属于敏感信息如果同一台机器上有多个业务系统最好把tokens.txt放在一个只有特定应用和运维账号可读的目录下文件权限设置600更合适。5. 工程化落地与后续扩展5.1 接入Jenkins定时刷新一旦你开始依赖token你会发现token是会过期的所以你需要一个定时刷新机制。Jenkins配合参数化构建是个非常成熟的方案。我在团队里是这样做的新建一个自由风格项目配置参数包括CSV文件路径、并发数、重试次数、输出文件路径。然后在构建步骤里执行python batch_login.py再在高级配置里勾选“如有失败记录则标红”。这样每次token快过期时手动触发一次构建或者用定时构建触发就能自动刷新tokens.txt。接入时要额外处理的一点是环境变量。不要把数据库密码、client_secret写进代码仓库Jenkins的凭据绑定插件可以把机密注入到环境变量中脚本再从环境变量里读取。这样做的好处是多环境之间只需要切换Jenkins凭据脚本本身不用改。跑完构建后我会在Post-build Actions里加一个Archive Artifacts步骤把output/tokens.txt归档到Jenkins构建产物目录这样如果下游系统坏了还能去历史构建记录里翻出上一批token应急。这个小小的“存档”习惯帮我们解决过不止一次线上问题强烈推荐。5.2 作为接口自动化测试的数据源tokens.txt最大的下游价值是当数据源。接口自动化测试框架里很多用例要求登录态才能访问业务接口以前的做法是每个用例自己登录一遍既慢又容易触发风控。现在你只需要在框架的conftest或者BaseTestCase里写一个读文件函数从tokens.txt里随机或者顺序取一个未过期的token塞进全局请求头即可。使用的时候要注意两个细节。第一个是token不能跨用户串着用尤其是涉及权限校验的接口某个用户的token去操作另一个用户的资源极容易被判定为越权。所以我建议读token的同时也要读取左边的用户名在断言里校验操作人身份。第二个细节是token池要预留一定余量比如你有1000个有效token不要让下游同时占用全部否则当某一条用例异常忘记释放整个token池就会被写满。如果你正在搭建AI自动化测试平台这个token池依旧适用。AI生成本身也需要接口鉴权批量获取token的能力可以直接内化为平台的数据准备服务对外提供接口比如启动压测前自动调用batch_login任务并等待完成。5.3 安全合规与超大规模扩展批量登录意味着你会持有大量用户的凭据和token安全责任一下子重了很多。我给自己定了几条铁律第一批量脚本只允许访问测试环境或自己拥有授权的环境严禁去扫第三方系统第二密码不落到日志里fetch_token函数里打日志时只打用户名和状态码绝不打印password第三tokens.txt文件权限至少设置为600并明确只允许特定应用账号读取。如果业务量从1000个用户膨胀到10000个甚至更多单机线程池就开始吃力了。这时候可以换成异步方案用aiohttp或者httpx的AsyncClient单机能承担更高的并发再往上就引入消息队列把10000个用户任务拆成多个分片每个分片由不同的worker处理最后再合并输出。分布式也不难Jenkins的多节点并行构建就能当简易任务分发器来用。还有一个方向是使用云函数做弹性扩缩容。每个云函数实例跑100个用户触发100个实例并行总时长可以压缩到几十秒以内。这种方式适合活动前临时大批量造数据的场景但成本和日志收集会复杂一些不建议一开始就上先把单机版跑稳再谈扩展。我个人在做这类批量任务时最深的体会是“稳定胜过速度”。你让脚本跑得快不难难的是跑完1000个用户之后你能清清楚楚说出哪些成功、哪些失败、失败在哪个环节、失败原因是什么。所以我建议你在写第一版脚本时就把日志、重试、失败名单、统计摘要这四样东西做进去以后遇到任何问题都会省心很多。最后再多说一个小技巧每次生成tokens.txt时顺手把当时的Python版本、requests版本、脚本版本号写进附属meta.json这样一旦后续行为变化你还能复盘是不是环境改动导致的结果差异。