拓冰网站定制 
1. 为什么数据库压测总在“脚本能跑、结果不敢信”之间反复横跳做 Python 压测 Mysql 和 Doris 这件事很多人第一次写脚本时都会觉得挺简单装个 pymysql开个线程池循环 insert 就完事了。但真正跑起来之后问题一个接一个冒出来——连接池被打爆、Doris 的 Stream Load 和普通 insert 行为不一致、压测跑完不知道数据到底写进去多少、想加个“让模型帮我分析压测结果”的环节又发现每个脚本都要单独配一套 Key。我自己最早压 Doris 的时候用的是最朴素的pymysql直连50 个线程每个插 20 条跑完打印一个总耗时就算完事。结果第一次跑就翻车连接数直接顶到上限报Too many connections第二次把连接池加上又发现 Doris 的写入延迟和 Mysql 完全不是一个量级用同一套参数去压两个库出来的数字根本没法横向对比。所以这篇要解决的不是“怎么写出一个能跑的压测脚本”而是怎么搭一套可复用、结果可校验、还能顺手接上大模型做结果分析的数据库压测方案。核心链路分四段第一段是压测脚本本身的设计包括连接池参数、并发模型、写入批次怎么定第二段是 Mysql 和 Doris 两套目标库的差异化配置因为这两个库的写入语义差别很大第三段是结果校验压测跑完必须能回答“到底成功写了多少条、失败的是哪些、耗时分布长什么样”第四段是把压测脚本和结果校验都接到一个统一的 Key 上这样你换模型、换分析脚本时不用到处改配置。这里说的“统一 Key”指的是用 TaoToken 这类聚合入口来管理模型调用凭证。压测脚本里如果需要调用模型做结果解读、生成压测报告、或者让模型帮你判断某次压测的 P99 是否异常都可以走同一个 Base URL 和同一个 Key不用在 Mysql 压测脚本、Doris 压测脚本、结果分析脚本里各维护一套。适合谁看已经会写基础 Python 脚本、但压测结果总是“看着像那么回事、细看又说不清”的后端或数据开发以及想把压测链路和 AI 辅助分析串起来、又不想在每个脚本里重复配 Key 的人。下面从环境准备开始一步步把这条链路搭起来。我会把 Mysql 和 Doris 的配置差异、连接池参数、并发模型、结果校验 SQL、以及 TaoToken 的接入方式都写成可直接复制的形式。你跟着做最后能拿到一个跑得稳、结果可对账、还能让模型帮你读报告的压测方案。2. TaoToken 统一 Key 接入让压测脚本和结果分析共用一套凭证在正式写压测脚本之前先把 TaoToken 这一层接进来。原因很简单如果你的压测链路里只有数据库操作那确实不需要模型但一旦你想加“压测结果自动解读”“异常耗时归因”“生成对比报告”这些环节就会涉及模型调用。与其等到后面再补不如一开始就把凭证层统一好。TaoToken 的定位是一个模型调用聚合入口官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。它的作用是让你用同一个 Base URL 和同一个 API Key去调用不同的模型而不需要在每个脚本里分别配置不同厂商的地址和密钥。对于压测场景来说这个统一 Key 的价值体现在三个地方第一压测脚本里如果需要模型参与比如让模型根据压测参数生成测试数据分布、或者根据历史压测结果推荐并发数可以直接在脚本里读同一个环境变量不用为每个脚本单独配。第二结果校验环节如果要做“智能对账”比如把 Mysql 和 Doris 的写入条数、耗时分布丢给模型做差异分析模型调用的凭证和压测脚本用的是同一套。第三你后续如果换模型比如从 A 模型换到 B 模型做报告生成只需要改模型 IDBase URL 和 Key 不用动。接入方式分两步。第一步是拿到 Key第二步是在压测项目里统一读取。拿 Key 的入口在控制台的 API Keys 页面地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。进去之后创建一个 Key复制出来。注意这个 Key 只显示一次建议直接写进项目的.env文件不要硬编码在脚本里。第二步是在压测项目根目录建一个.env文件内容如下# .env TAOTOKEN_API_KEYsk-你的实际key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_IDclaude-3-5-sonnet然后在 Python 里统一读取。我习惯用一个config.py来管理这样压测脚本、结果校验脚本、报告生成脚本都从这里拿配置# config.py import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY) TAOTOKEN_BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) TAOTOKEN_MODEL_ID os.getenv(TAOTOKEN_MODEL_ID, claude-3-5-sonnet) # 数据库配置也统一放这里 MYSQL_CONFIG { host: os.getenv(MYSQL_HOST, 127.0.0.1), port: int(os.getenv(MYSQL_PORT, 3306)), user: os.getenv(MYSQL_USER, root), password: os.getenv(MYSQL_PASSWORD, ), database: os.getenv(MYSQL_DB, test), } DORIS_CONFIG { host: os.getenv(DORIS_HOST, 127.0.0.1), port: int(os.getenv(DORIS_PORT, 9030)), user: os.getenv(DORIS_USER, root), password: os.getenv(DORIS_PASSWORD, ), database: os.getenv(DORIS_DB, test), }这里有个细节要注意Doris 的 FE 查询端口默认是 9030BE 的 HTTP 端口是 8040Stream Load 走的是 8030。如果你用 pymysql 直连 Doris连的是 9030走的是 MySQL 协议。这一点和 Mysql 的 3306 不一样配置里要区分开。如果你用的是 Claude Code 这类编码工具来辅助写压测脚本可以在工具里配置 Anthropic 兼容的 Base URL。Claude Code 的配置入口在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite 里面会告诉你 Base URL 和 Key 怎么填。配置好之后你在写压测脚本时让工具帮你补全连接池参数、生成校验 SQL都会走这个统一入口。模型对话的调试入口在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 你可以先在那里试一下 Key 能不能正常调通再写进脚本。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有不同语言调用方式的说明。Python 侧用 OpenAI SDK 或 Anthropic SDK 都可以取决于你选的模型。这一层配好之后后面压测脚本里如果需要模型参与直接from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL, TAOTOKEN_MODEL_ID就行。不用在每个脚本里重复写地址和密钥换模型也只改.env一个地方。有一点要提醒TaoToken 是模型调用入口不是数据库代理。压测脚本连 Mysql 和 Doris 走的还是各自的数据库连接TaoToken 只负责模型调用那一层。不要把数据库连接也往 TaoToken 上套那是两回事。3. 可复制配置Mysql 与 Doris 压测脚本的连接池、并发与写入参数这一节是整篇的核心给出可直接复制的压测脚本配置。我会把 Mysql 和 Doris 分开写因为这两个库在写入行为上有本质差异用同一套参数去压会得出误导性结论。先看 Mysql 的压测脚本。Mysql 的写入是行级事务每次commit都会落盘所以并发模型和批次大小对吞吐影响很大。连接池用dbutils的PooledDB参数如下# mysql_bench.py import time import pymysql from dbutils.pooled_db import PooledDB from concurrent.futures import ThreadPoolExecutor, as_completed from config import MYSQL_CONFIG MYSQL_POOL PooledDB( creatorpymysql, hostMYSQL_CONFIG[host], portMYSQL_CONFIG[port], userMYSQL_CONFIG[user], passwordMYSQL_CONFIG[password], databaseMYSQL_CONFIG[database], charsetutf8mb4, connect_timeout10, read_timeout30, write_timeout30, maxconnections100, mincached5, maxcached20, maxshared10, blockingTrue, cursorclasspymysql.cursors.DictCursor, ) def mysql_insert_batch(thread_id, batch_size, tabletbl_point_query1): conn MYSQL_POOL.connection() success 0 fail 0 latencies [] try: with conn.cursor() as cursor: for i in range(batch_size): data_id thread_id * batch_size i 1 start time.time() try: cursor.execute( fINSERT INTO {table} (key, table_name, column_name, data_type) fVALUES (%s, %s, %s, %s), (data_id, value1, value2, value3), ) conn.commit() success 1 except Exception as e: conn.rollback() fail 1 latencies.append(time.time() - start) finally: conn.close() return {thread_id: thread_id, success: success, fail: fail, latencies: latencies} def run_mysql_bench(threads50, batch_size20): start time.time() results [] with ThreadPoolExecutor(max_workersthreads) as executor: futures [executor.submit(mysql_insert_batch, i, batch_size) for i in range(threads)] for future in as_completed(futures): results.append(future.result()) total_time time.time() - start total_success sum(r[success] for r in results) total_fail sum(r[fail] for r in results) all_latencies [l for r in results for l in r[latencies]] all_latencies.sort() p50 all_latencies[len(all_latencies) // 2] if all_latencies else 0 p99 all_latencies[int(len(all_latencies) * 0.99)] if all_latencies else 0 return { total_time: round(total_time, 3), total_success: total_success, total_fail: total_fail, qps: round(total_success / total_time, 2) if total_time 0 else 0, p50: round(p50, 4), p99: round(p99, 4), } if __name__ __main__: print(run_mysql_bench(threads50, batch_size20))这段脚本和原始 excerpt 的区别在于原始版本用 f-string 直接拼 SQL有注入风险且无法复用这里改成参数化查询。原始版本没有统计成功/失败条数压测跑完只知道“插了”不知道“插进去多少”这里每个线程返回 success 和 fail最后汇总。原始版本没有延迟分布这里记录了每条 insert 的耗时最后算 P50 和 P99。再看 Doris 的压测脚本。Doris 走 MySQL 协议时INSERT是同步写入但 Doris 的写入路径和 Mysql 完全不同——它先写 WAL再异步刷 BE。所以用同样的线程数和批次去压Doris 的延迟通常比 Mysql 高但吞吐可能更稳。Doris 的连接池配置要单独调# doris_bench.py import time import pymysql from dbutils.pooled_db import PooledDB from concurrent.futures import ThreadPoolExecutor, as_completed from config import DORIS_CONFIG DORIS_POOL PooledDB( creatorpymysql, hostDORIS_CONFIG[host], portDORIS_CONFIG[port], userDORIS_CONFIG[user], passwordDORIS_CONFIG[password], databaseDORIS_CONFIG[database], charsetutf8mb4, connect_timeout10, read_timeout60, write_timeout60, maxconnections80, mincached5, maxcached15, maxshared5, blockingTrue, cursorclasspymysql.cursors.DictCursor, ) def doris_insert_batch(thread_id, batch_size, tabletbl_point_query1): conn DORIS_POOL.connection() success 0 fail 0 latencies [] try: with conn.cursor() as cursor: for i in range(batch_size): data_id thread_id * batch_size i 1 start time.time() try: cursor.execute( fINSERT INTO {table} (key, table_name, column_name, data_type) fVALUES (%s, %s, %s, %s), (data_id, value1, value2, value3), ) success 1 except Exception as e: fail 1 latencies.append(time.time() - start) finally: conn.close() return {thread_id: thread_id, success: success, fail: fail, latencies: latencies} def run_doris_bench(threads40, batch_size25): start time.time() results [] with ThreadPoolExecutor(max_workersthreads) as executor: futures [executor.submit(doris_insert_batch, i, batch_size) for i in range(threads)] for future in as_completed(futures): results.append(future.result()) total_time time.time() - start total_success sum(r[success] for r in results) total_fail sum(r[fail] for r in results) all_latencies [l for r in results for l in r[latencies]] all_latencies.sort() p50 all_latencies[len(all_latencies) // 2] if all_latencies else 0 p99 all_latencies[int(len(all_latencies) * 0.99)] if all_latencies else 0 return { total_time: round(total_time, 3), total_success: total_success, total_fail: total_fail, qps: round(total_success / total_time, 2) if total_time 0 else 0, p50: round(p50, 4), p99: round(p99, 4), } if __name__ __main__: print(run_doris_bench(threads40, batch_size25))两个脚本的关键差异我列成表格方便对照参数MysqlDoris说明端口33069030Doris 走 FE 的 MySQL 协议端口maxconnections10080Doris 连接开销更大不宜开太高read_timeout3060Doris 写入延迟更高超时要放宽write_timeout3060同上默认线程数5040Doris 并发过高反而会触发限流默认批次2025Doris 单条写入开销大批次可略大commit 行为每条 commit每条自动提交Doris 不支持显式事务回滚这里要特别说 Doris 的 commit 行为。Doris 走 MySQL 协议时INSERT是自动提交的你调conn.commit()不会报错但也没有实际事务语义。所以 Doris 脚本里失败重试要谨慎不能像 Mysql 那样 rollback 后重插否则可能产生重复数据。我的做法是 Doris 侧失败就记录不自动重试压测结束后统一对账。如果你想让模型帮你根据压测目标推荐参数比如“我要压 5000 QPS线程数和批次怎么配”可以在脚本里加一个调用 TaoToken 的函数# llm_helper.py from openai import OpenAI from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL, TAOTOKEN_MODEL_ID client OpenAI(api_keyTAOTOKEN_API_KEY, base_urlTAOTOKEN_BASE_URL) def suggest_bench_params(target_qps, db_typemysql): prompt f我要对 {db_type} 做压测目标 QPS 是 {target_qps}请给出线程数和批次大小的建议并说明理由。 resp client.chat.completions.create( modelTAOTOKEN_MODEL_ID, messages[{role: user, content: prompt}], temperature0.3, ) return resp.choices[0].message.content这个函数用的是 OpenAI SDK 的调用方式Base URL 指向 TaoToken 的 API 入口。你换成 Anthropic SDK 也可以取决于.env里配的模型。这样压测脚本和模型调用共用一套凭证不用额外配。配置部分到这里就齐了。Mysql 和 Doris 各一套连接池参数、并发模型、写入逻辑加上一个可选的模型辅助函数。下一节讲怎么验证这些配置真的跑通了以及结果怎么对账。4. 验证请求与成功结果压测跑通后怎么确认数据真的写进去了压测脚本能跑完不代表数据写对了。我见过太多次“脚本打印 All data inserted去数据库一查条数对不上”的情况。所以这一节讲两件事一是怎么验证压测请求本身是通的二是怎么校验写入结果的一致性。先验证请求通路。最直接的方式是跑一个小批量比如 2 个线程各插 5 条然后去数据库查。Mysql 侧SELECT COUNT(*) FROM test.tbl_point_query1; SELECT * FROM test.tbl_point_query1 ORDER BY key DESC LIMIT 10;Doris 侧同样的 SQL但要注意 Doris 的查询延迟可能比 Mysql 高刚写完立刻查可能因为 compaction 还没完成而看到旧数据。等几秒再查SELECT COUNT(*) FROM test.tbl_point_query1;如果条数对得上说明写入通路没问题。如果对不上先看压测脚本返回的 success 和 fail 计数。success 是脚本认为成功的条数数据库实际条数是 ground truth两者不一致通常有三种原因第一种是 Doris 的写入可见性延迟。Doris 的 INSERT 返回成功只代表 FE 接受了请求BE 实际落盘和可见可能有秒级延迟。这种情况等几秒再查就能对上。第二种是连接池配置问题。如果maxconnections设得太小blockingTrue会让线程排队等连接压测耗时里包含了等待时间但 success 计数还是准的。如果blockingFalse拿不到连接会直接抛异常这时候 fail 计数会上去。第三种是主键冲突或表结构不匹配。比如key字段是主键多个线程生成的 data_id 有重叠就会插入失败。原始 excerpt 里用thread_id * count i 1生成 ID如果线程数和批次变了ID 生成逻辑要跟着改否则会撞主键。验证请求通路的另一个方式是看压测脚本的输出结构。跑一次小批量输出应该类似{ total_time: 1.234, total_success: 10, total_fail: 0, qps: 8.1, p50: 0.0123, p99: 0.0456 }如果total_fail大于 0先别急着加线程先把失败原因打出来。可以在except里加一行print(fthread {thread_id} insert failed: {e})跑一次看报什么错。常见的是Duplicate entry或Table doesnt exist。请求通路验证完之后做结果一致性校验。这一步是压测方案里最容易被忽略的。我的做法是压测前后各查一次条数差值应该等于total_success# verify.py import pymysql from config import MYSQL_CONFIG, DORIS_CONFIG def count_rows(config, tabletbl_point_query1): conn pymysql.connect(**config, cursorclasspymysql.cursors.DictCursor) try: with conn.cursor() as cursor: cursor.execute(fSELECT COUNT(*) AS cnt FROM {table}) return cursor.fetchone()[cnt] finally: conn.close() def verify_bench(before_mysql, before_doris, mysql_result, doris_result): after_mysql count_rows(MYSQL_CONFIG) after_doris count_rows(DORIS_CONFIG) mysql_delta after_mysql - before_mysql doris_delta after_doris - before_doris print(fMysql 预期写入: {mysql_result[total_success]}, 实际增量: {mysql_delta}) print(fDoris 预期写入: {doris_result[total_success]}, 实际增量: {doris_delta}) if mysql_delta ! mysql_result[total_success]: print(Mysql 写入不一致需要排查) if doris_delta ! doris_result[total_success]: print(Doris 写入不一致可能还在 compaction稍后重查)跑完压测后调用这个校验输出类似Mysql 预期写入: 1000, 实际增量: 1000 Doris 预期写入: 1000, 实际增量: 998 Doris 写入不一致可能还在 compaction稍后重查Doris 差 2 条这种情况等 10 秒再查通常就一致了。如果一直不一致就要去看 BE 日志可能是某个 BE 节点写入失败但 FE 没返回错误。如果你想让模型帮你解读这个校验结果可以把mysql_result和doris_result拼成 prompt 丢给 TaoTokendef analyze_result(mysql_result, doris_result): prompt f 以下是 Mysql 和 Doris 的压测结果请分析差异并给出可能原因 Mysql: {mysql_result} Doris: {doris_result} resp client.chat.completions.create( modelTAOTOKEN_MODEL_ID, messages[{role: user, content: prompt}], temperature0.2, ) return resp.choices[0].message.content这样压测、校验、分析三段都走同一个 Key不用分别配。验证环节的核心就一句话脚本说成功不算成功数据库条数对得上才算。每次压测跑完先看 success/fail再查前后条数差两个都对上这次压测结果才可信。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 报错怎么定位压测链路里报错分两类数据库侧的报错和模型调用侧的报错。数据库侧的报错相对直观模型调用侧的报错因为多了一层网络和鉴权排查起来容易绕弯路。这一节把常见的几类报错和定位方法列出来。第一类数据库连接报错pymysql.err.OperationalError: (2003, Cant connect to MySQL server on 127.0.0.1)这种是连不上。先确认端口对不对Mysql 是 3306Doris 是 9030。再确认服务有没有起telnet 127.0.0.1 9030试一下。如果 Doris 的 FE 起了但 BE 没起连接能建立但写入会失败报错通常是Failed to find backend。pymysql.err.OperationalError: (1040, Too many connections)是连接数打满。把maxconnections调小或者把blocking设为 True 让线程排队。Doris 侧尤其要注意FE 的连接数上限默认不高压测线程别开太多。pymysql.err.IntegrityError: (1062, Duplicate entry xxx for key PRIMARY)是主键冲突。检查 data_id 生成逻辑确保多线程之间不重叠。原始 excerpt 里thread_id * count i 1这个公式如果count和实际批次不一致就会撞。第二类模型调用报错401 Unauthorized是最常见的。先检查.env里的TAOTOKEN_API_KEY有没有写错有没有多余空格。然后确认 Base URL 是不是https://taotoken.net/api注意结尾不要多加/v1SDK 会自己拼。如果用的是 Anthropic SDKBase URL 的拼法可能不同参考接入文档里的说明。local proxy failed这种报错通常和本地网络环境有关。先确认你的请求是直接发到 TaoToken 的 API 入口没有经过额外的本地转发。检查.env里有没有残留的HTTP_PROXY或HTTPS_PROXY环境变量有的话清掉。Python 里可以用os.environ.pop(HTTP_PROXY, None)在脚本开头清一下。Error reading choices或reading choices这类报错通常是响应体解析失败。原因可能是模型返回了非预期格式或者请求被截断。先确认TAOTOKEN_MODEL_ID填的模型名是存在的去模型对话页面确认一下。然后把temperature调低max_tokens设大一点避免返回被截断。OAuth相关报错如果你用的是 Claude Code 这类工具可能是工具的认证方式和 API Key 方式冲突了。Claude Code 的配置入口在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite 按里面的说明配 Base URL 和 Key。如果工具里同时开了 OAuth 和 API Key可能会互相覆盖建议只保留一种。第三类压测结果异常压测跑完 QPS 特别低先看total_time里有没有包含连接池等待时间。如果blockingTrue且maxconnections太小线程大部分时间在等连接QPS 自然上不去。把maxconnections调到和线程数匹配。P99 特别高但 P50 正常通常是偶发的连接重建或 Doris compaction。看压测期间有没有网络抖动或者 Doris 是不是在后台做 compaction。这种情况多跑几次取稳定值。total_fail大于 0 但数据库条数对得上可能是失败后重试成功了。检查脚本里有没有自动重试逻辑如果有fail 计数和实际失败数会不一致。建议失败就记录不自动重试压测结束后统一对账。第四类配置类报错KeyError: TAOTOKEN_API_KEY是.env没加载。确认python-dotenv装了load_dotenv()在读取环境变量之前调用了。如果.env文件和脚本不在同一目录load_dotenv()要传路径。ModuleNotFoundError: No module named dbutils是依赖没装。pip install dbutils pymysql python-dotenv openai一把装齐。Doris 侧如果报Unknown table tbl_point_query1先确认表建了没有库名对不对。Doris 的库表是大小写敏感的test.tbl_point_query1和test.TBL_POINT_QUERY1是两张表。排查的核心思路是先分清是数据库侧还是模型侧再看是连接问题还是数据问题最后看是配置问题还是代码问题。每一类报错都有对应的检查点按顺序过一遍大部分问题都能定位到。6. 把压测、校验、分析串成一条可复用的链路到这里Mysql 和 Doris 的压测脚本、连接池配置、结果校验、模型调用都已经能跑通了。最后说一下怎么把这些串成一条可复用的链路而不是每次压测都手动拼。我的做法是建一个bench_runner.py把压测、校验、分析三步串起来# bench_runner.py from mysql_bench import run_mysql_bench from doris_bench import run_doris_bench from verify import count_rows, verify_bench from llm_helper import analyze_result from config import MYSQL_CONFIG, DORIS_CONFIG def run_full_bench(mysql_threads50, mysql_batch20, doris_threads40, doris_batch25): before_mysql count_rows(MYSQL_CONFIG) before_doris count_rows(DORIS_CONFIG) mysql_result run_mysql_bench(mysql_threads, mysql_batch) doris_result run_doris_bench(doris_threads, doris_batch) verify_bench(before_mysql, before_doris, mysql_result, doris_result) analysis analyze_result(mysql_result, doris_result) print(模型分析结果) print(analysis) return {mysql: mysql_result, doris: doris_result, analysis: analysis} if __name__ __main__: run_full_bench()这样每次压测只需要调run_full_bench()传入线程数和批次剩下的校验和分析自动完成。模型分析那一步走的是 TaoToken 的统一 Key和压测脚本共用.env里的配置。如果你要长期做压测建议把每次结果存下来方便对比。可以存成 JSON字段包括时间戳、线程数、批次、QPS、P50、P99、success、fail。下次压测时把历史结果一起丢给模型让模型帮你判断这次是变好了还是变差了。Coding Plan 的入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 如果你要长期跑压测和 Agent 类任务可以看一下那边的方案适合需要持续调用模型的场景。最后说一个我踩过的坑Doris 压测时不要用太高的并发去压单 BE 节点BE 的写入线程池有限并发过高会排队QPS 反而下降。我试过把线程从 40 加到 80QPS 没涨P99 翻了一倍。后来降到 40QPS 反而更稳。所以压测参数不是越大越好找到拐点比堆线程更重要。整套链路的核心就三件事压测脚本要能区分 Mysql 和 Doris 的写入语义结果校验要以数据库实际条数为准模型调用要统一 Key 避免到处配。这三件做好压测结果才敢拿去用。 Python 压测 Mysql、Doris 实战：用 TaoToken 统一 Key 打通压测脚本与结果校验
1. 为什么数据库压测总在“脚本能跑、结果不敢信”之间反复横跳做 Python 压测 Mysql 和 Doris 这件事很多人第一次写脚本时都会觉得挺简单装个 pymysql开个线程池循环 insert 就完事了。但真正跑起来之后问题一个接一个冒出来——连接池被打爆、Doris 的 Stream Load 和普通 insert 行为不一致、压测跑完不知道数据到底写进去多少、想加个“让模型帮我分析压测结果”的环节又发现每个脚本都要单独配一套 Key。我自己最早压 Doris 的时候用的是最朴素的pymysql直连50 个线程每个插 20 条跑完打印一个总耗时就算完事。结果第一次跑就翻车连接数直接顶到上限报Too many connections第二次把连接池加上又发现 Doris 的写入延迟和 Mysql 完全不是一个量级用同一套参数去压两个库出来的数字根本没法横向对比。所以这篇要解决的不是“怎么写出一个能跑的压测脚本”而是怎么搭一套可复用、结果可校验、还能顺手接上大模型做结果分析的数据库压测方案。核心链路分四段第一段是压测脚本本身的设计包括连接池参数、并发模型、写入批次怎么定第二段是 Mysql 和 Doris 两套目标库的差异化配置因为这两个库的写入语义差别很大第三段是结果校验压测跑完必须能回答“到底成功写了多少条、失败的是哪些、耗时分布长什么样”第四段是把压测脚本和结果校验都接到一个统一的 Key 上这样你换模型、换分析脚本时不用到处改配置。这里说的“统一 Key”指的是用 TaoToken 这类聚合入口来管理模型调用凭证。压测脚本里如果需要调用模型做结果解读、生成压测报告、或者让模型帮你判断某次压测的 P99 是否异常都可以走同一个 Base URL 和同一个 Key不用在 Mysql 压测脚本、Doris 压测脚本、结果分析脚本里各维护一套。适合谁看已经会写基础 Python 脚本、但压测结果总是“看着像那么回事、细看又说不清”的后端或数据开发以及想把压测链路和 AI 辅助分析串起来、又不想在每个脚本里重复配 Key 的人。下面从环境准备开始一步步把这条链路搭起来。我会把 Mysql 和 Doris 的配置差异、连接池参数、并发模型、结果校验 SQL、以及 TaoToken 的接入方式都写成可直接复制的形式。你跟着做最后能拿到一个跑得稳、结果可对账、还能让模型帮你读报告的压测方案。2. TaoToken 统一 Key 接入让压测脚本和结果分析共用一套凭证在正式写压测脚本之前先把 TaoToken 这一层接进来。原因很简单如果你的压测链路里只有数据库操作那确实不需要模型但一旦你想加“压测结果自动解读”“异常耗时归因”“生成对比报告”这些环节就会涉及模型调用。与其等到后面再补不如一开始就把凭证层统一好。TaoToken 的定位是一个模型调用聚合入口官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。它的作用是让你用同一个 Base URL 和同一个 API Key去调用不同的模型而不需要在每个脚本里分别配置不同厂商的地址和密钥。对于压测场景来说这个统一 Key 的价值体现在三个地方第一压测脚本里如果需要模型参与比如让模型根据压测参数生成测试数据分布、或者根据历史压测结果推荐并发数可以直接在脚本里读同一个环境变量不用为每个脚本单独配。第二结果校验环节如果要做“智能对账”比如把 Mysql 和 Doris 的写入条数、耗时分布丢给模型做差异分析模型调用的凭证和压测脚本用的是同一套。第三你后续如果换模型比如从 A 模型换到 B 模型做报告生成只需要改模型 IDBase URL 和 Key 不用动。接入方式分两步。第一步是拿到 Key第二步是在压测项目里统一读取。拿 Key 的入口在控制台的 API Keys 页面地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。进去之后创建一个 Key复制出来。注意这个 Key 只显示一次建议直接写进项目的.env文件不要硬编码在脚本里。第二步是在压测项目根目录建一个.env文件内容如下# .env TAOTOKEN_API_KEYsk-你的实际key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_IDclaude-3-5-sonnet然后在 Python 里统一读取。我习惯用一个config.py来管理这样压测脚本、结果校验脚本、报告生成脚本都从这里拿配置# config.py import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY) TAOTOKEN_BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) TAOTOKEN_MODEL_ID os.getenv(TAOTOKEN_MODEL_ID, claude-3-5-sonnet) # 数据库配置也统一放这里 MYSQL_CONFIG { host: os.getenv(MYSQL_HOST, 127.0.0.1), port: int(os.getenv(MYSQL_PORT, 3306)), user: os.getenv(MYSQL_USER, root), password: os.getenv(MYSQL_PASSWORD, ), database: os.getenv(MYSQL_DB, test), } DORIS_CONFIG { host: os.getenv(DORIS_HOST, 127.0.0.1), port: int(os.getenv(DORIS_PORT, 9030)), user: os.getenv(DORIS_USER, root), password: os.getenv(DORIS_PASSWORD, ), database: os.getenv(DORIS_DB, test), }这里有个细节要注意Doris 的 FE 查询端口默认是 9030BE 的 HTTP 端口是 8040Stream Load 走的是 8030。如果你用 pymysql 直连 Doris连的是 9030走的是 MySQL 协议。这一点和 Mysql 的 3306 不一样配置里要区分开。如果你用的是 Claude Code 这类编码工具来辅助写压测脚本可以在工具里配置 Anthropic 兼容的 Base URL。Claude Code 的配置入口在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite 里面会告诉你 Base URL 和 Key 怎么填。配置好之后你在写压测脚本时让工具帮你补全连接池参数、生成校验 SQL都会走这个统一入口。模型对话的调试入口在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 你可以先在那里试一下 Key 能不能正常调通再写进脚本。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有不同语言调用方式的说明。Python 侧用 OpenAI SDK 或 Anthropic SDK 都可以取决于你选的模型。这一层配好之后后面压测脚本里如果需要模型参与直接from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL, TAOTOKEN_MODEL_ID就行。不用在每个脚本里重复写地址和密钥换模型也只改.env一个地方。有一点要提醒TaoToken 是模型调用入口不是数据库代理。压测脚本连 Mysql 和 Doris 走的还是各自的数据库连接TaoToken 只负责模型调用那一层。不要把数据库连接也往 TaoToken 上套那是两回事。3. 可复制配置Mysql 与 Doris 压测脚本的连接池、并发与写入参数这一节是整篇的核心给出可直接复制的压测脚本配置。我会把 Mysql 和 Doris 分开写因为这两个库在写入行为上有本质差异用同一套参数去压会得出误导性结论。先看 Mysql 的压测脚本。Mysql 的写入是行级事务每次commit都会落盘所以并发模型和批次大小对吞吐影响很大。连接池用dbutils的PooledDB参数如下# mysql_bench.py import time import pymysql from dbutils.pooled_db import PooledDB from concurrent.futures import ThreadPoolExecutor, as_completed from config import MYSQL_CONFIG MYSQL_POOL PooledDB( creatorpymysql, hostMYSQL_CONFIG[host], portMYSQL_CONFIG[port], userMYSQL_CONFIG[user], passwordMYSQL_CONFIG[password], databaseMYSQL_CONFIG[database], charsetutf8mb4, connect_timeout10, read_timeout30, write_timeout30, maxconnections100, mincached5, maxcached20, maxshared10, blockingTrue, cursorclasspymysql.cursors.DictCursor, ) def mysql_insert_batch(thread_id, batch_size, tabletbl_point_query1): conn MYSQL_POOL.connection() success 0 fail 0 latencies [] try: with conn.cursor() as cursor: for i in range(batch_size): data_id thread_id * batch_size i 1 start time.time() try: cursor.execute( fINSERT INTO {table} (key, table_name, column_name, data_type) fVALUES (%s, %s, %s, %s), (data_id, value1, value2, value3), ) conn.commit() success 1 except Exception as e: conn.rollback() fail 1 latencies.append(time.time() - start) finally: conn.close() return {thread_id: thread_id, success: success, fail: fail, latencies: latencies} def run_mysql_bench(threads50, batch_size20): start time.time() results [] with ThreadPoolExecutor(max_workersthreads) as executor: futures [executor.submit(mysql_insert_batch, i, batch_size) for i in range(threads)] for future in as_completed(futures): results.append(future.result()) total_time time.time() - start total_success sum(r[success] for r in results) total_fail sum(r[fail] for r in results) all_latencies [l for r in results for l in r[latencies]] all_latencies.sort() p50 all_latencies[len(all_latencies) // 2] if all_latencies else 0 p99 all_latencies[int(len(all_latencies) * 0.99)] if all_latencies else 0 return { total_time: round(total_time, 3), total_success: total_success, total_fail: total_fail, qps: round(total_success / total_time, 2) if total_time 0 else 0, p50: round(p50, 4), p99: round(p99, 4), } if __name__ __main__: print(run_mysql_bench(threads50, batch_size20))这段脚本和原始 excerpt 的区别在于原始版本用 f-string 直接拼 SQL有注入风险且无法复用这里改成参数化查询。原始版本没有统计成功/失败条数压测跑完只知道“插了”不知道“插进去多少”这里每个线程返回 success 和 fail最后汇总。原始版本没有延迟分布这里记录了每条 insert 的耗时最后算 P50 和 P99。再看 Doris 的压测脚本。Doris 走 MySQL 协议时INSERT是同步写入但 Doris 的写入路径和 Mysql 完全不同——它先写 WAL再异步刷 BE。所以用同样的线程数和批次去压Doris 的延迟通常比 Mysql 高但吞吐可能更稳。Doris 的连接池配置要单独调# doris_bench.py import time import pymysql from dbutils.pooled_db import PooledDB from concurrent.futures import ThreadPoolExecutor, as_completed from config import DORIS_CONFIG DORIS_POOL PooledDB( creatorpymysql, hostDORIS_CONFIG[host], portDORIS_CONFIG[port], userDORIS_CONFIG[user], passwordDORIS_CONFIG[password], databaseDORIS_CONFIG[database], charsetutf8mb4, connect_timeout10, read_timeout60, write_timeout60, maxconnections80, mincached5, maxcached15, maxshared5, blockingTrue, cursorclasspymysql.cursors.DictCursor, ) def doris_insert_batch(thread_id, batch_size, tabletbl_point_query1): conn DORIS_POOL.connection() success 0 fail 0 latencies [] try: with conn.cursor() as cursor: for i in range(batch_size): data_id thread_id * batch_size i 1 start time.time() try: cursor.execute( fINSERT INTO {table} (key, table_name, column_name, data_type) fVALUES (%s, %s, %s, %s), (data_id, value1, value2, value3), ) success 1 except Exception as e: fail 1 latencies.append(time.time() - start) finally: conn.close() return {thread_id: thread_id, success: success, fail: fail, latencies: latencies} def run_doris_bench(threads40, batch_size25): start time.time() results [] with ThreadPoolExecutor(max_workersthreads) as executor: futures [executor.submit(doris_insert_batch, i, batch_size) for i in range(threads)] for future in as_completed(futures): results.append(future.result()) total_time time.time() - start total_success sum(r[success] for r in results) total_fail sum(r[fail] for r in results) all_latencies [l for r in results for l in r[latencies]] all_latencies.sort() p50 all_latencies[len(all_latencies) // 2] if all_latencies else 0 p99 all_latencies[int(len(all_latencies) * 0.99)] if all_latencies else 0 return { total_time: round(total_time, 3), total_success: total_success, total_fail: total_fail, qps: round(total_success / total_time, 2) if total_time 0 else 0, p50: round(p50, 4), p99: round(p99, 4), } if __name__ __main__: print(run_doris_bench(threads40, batch_size25))两个脚本的关键差异我列成表格方便对照参数MysqlDoris说明端口33069030Doris 走 FE 的 MySQL 协议端口maxconnections10080Doris 连接开销更大不宜开太高read_timeout3060Doris 写入延迟更高超时要放宽write_timeout3060同上默认线程数5040Doris 并发过高反而会触发限流默认批次2025Doris 单条写入开销大批次可略大commit 行为每条 commit每条自动提交Doris 不支持显式事务回滚这里要特别说 Doris 的 commit 行为。Doris 走 MySQL 协议时INSERT是自动提交的你调conn.commit()不会报错但也没有实际事务语义。所以 Doris 脚本里失败重试要谨慎不能像 Mysql 那样 rollback 后重插否则可能产生重复数据。我的做法是 Doris 侧失败就记录不自动重试压测结束后统一对账。如果你想让模型帮你根据压测目标推荐参数比如“我要压 5000 QPS线程数和批次怎么配”可以在脚本里加一个调用 TaoToken 的函数# llm_helper.py from openai import OpenAI from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL, TAOTOKEN_MODEL_ID client OpenAI(api_keyTAOTOKEN_API_KEY, base_urlTAOTOKEN_BASE_URL) def suggest_bench_params(target_qps, db_typemysql): prompt f我要对 {db_type} 做压测目标 QPS 是 {target_qps}请给出线程数和批次大小的建议并说明理由。 resp client.chat.completions.create( modelTAOTOKEN_MODEL_ID, messages[{role: user, content: prompt}], temperature0.3, ) return resp.choices[0].message.content这个函数用的是 OpenAI SDK 的调用方式Base URL 指向 TaoToken 的 API 入口。你换成 Anthropic SDK 也可以取决于.env里配的模型。这样压测脚本和模型调用共用一套凭证不用额外配。配置部分到这里就齐了。Mysql 和 Doris 各一套连接池参数、并发模型、写入逻辑加上一个可选的模型辅助函数。下一节讲怎么验证这些配置真的跑通了以及结果怎么对账。4. 验证请求与成功结果压测跑通后怎么确认数据真的写进去了压测脚本能跑完不代表数据写对了。我见过太多次“脚本打印 All data inserted去数据库一查条数对不上”的情况。所以这一节讲两件事一是怎么验证压测请求本身是通的二是怎么校验写入结果的一致性。先验证请求通路。最直接的方式是跑一个小批量比如 2 个线程各插 5 条然后去数据库查。Mysql 侧SELECT COUNT(*) FROM test.tbl_point_query1; SELECT * FROM test.tbl_point_query1 ORDER BY key DESC LIMIT 10;Doris 侧同样的 SQL但要注意 Doris 的查询延迟可能比 Mysql 高刚写完立刻查可能因为 compaction 还没完成而看到旧数据。等几秒再查SELECT COUNT(*) FROM test.tbl_point_query1;如果条数对得上说明写入通路没问题。如果对不上先看压测脚本返回的 success 和 fail 计数。success 是脚本认为成功的条数数据库实际条数是 ground truth两者不一致通常有三种原因第一种是 Doris 的写入可见性延迟。Doris 的 INSERT 返回成功只代表 FE 接受了请求BE 实际落盘和可见可能有秒级延迟。这种情况等几秒再查就能对上。第二种是连接池配置问题。如果maxconnections设得太小blockingTrue会让线程排队等连接压测耗时里包含了等待时间但 success 计数还是准的。如果blockingFalse拿不到连接会直接抛异常这时候 fail 计数会上去。第三种是主键冲突或表结构不匹配。比如key字段是主键多个线程生成的 data_id 有重叠就会插入失败。原始 excerpt 里用thread_id * count i 1生成 ID如果线程数和批次变了ID 生成逻辑要跟着改否则会撞主键。验证请求通路的另一个方式是看压测脚本的输出结构。跑一次小批量输出应该类似{ total_time: 1.234, total_success: 10, total_fail: 0, qps: 8.1, p50: 0.0123, p99: 0.0456 }如果total_fail大于 0先别急着加线程先把失败原因打出来。可以在except里加一行print(fthread {thread_id} insert failed: {e})跑一次看报什么错。常见的是Duplicate entry或Table doesnt exist。请求通路验证完之后做结果一致性校验。这一步是压测方案里最容易被忽略的。我的做法是压测前后各查一次条数差值应该等于total_success# verify.py import pymysql from config import MYSQL_CONFIG, DORIS_CONFIG def count_rows(config, tabletbl_point_query1): conn pymysql.connect(**config, cursorclasspymysql.cursors.DictCursor) try: with conn.cursor() as cursor: cursor.execute(fSELECT COUNT(*) AS cnt FROM {table}) return cursor.fetchone()[cnt] finally: conn.close() def verify_bench(before_mysql, before_doris, mysql_result, doris_result): after_mysql count_rows(MYSQL_CONFIG) after_doris count_rows(DORIS_CONFIG) mysql_delta after_mysql - before_mysql doris_delta after_doris - before_doris print(fMysql 预期写入: {mysql_result[total_success]}, 实际增量: {mysql_delta}) print(fDoris 预期写入: {doris_result[total_success]}, 实际增量: {doris_delta}) if mysql_delta ! mysql_result[total_success]: print(Mysql 写入不一致需要排查) if doris_delta ! doris_result[total_success]: print(Doris 写入不一致可能还在 compaction稍后重查)跑完压测后调用这个校验输出类似Mysql 预期写入: 1000, 实际增量: 1000 Doris 预期写入: 1000, 实际增量: 998 Doris 写入不一致可能还在 compaction稍后重查Doris 差 2 条这种情况等 10 秒再查通常就一致了。如果一直不一致就要去看 BE 日志可能是某个 BE 节点写入失败但 FE 没返回错误。如果你想让模型帮你解读这个校验结果可以把mysql_result和doris_result拼成 prompt 丢给 TaoTokendef analyze_result(mysql_result, doris_result): prompt f 以下是 Mysql 和 Doris 的压测结果请分析差异并给出可能原因 Mysql: {mysql_result} Doris: {doris_result} resp client.chat.completions.create( modelTAOTOKEN_MODEL_ID, messages[{role: user, content: prompt}], temperature0.2, ) return resp.choices[0].message.content这样压测、校验、分析三段都走同一个 Key不用分别配。验证环节的核心就一句话脚本说成功不算成功数据库条数对得上才算。每次压测跑完先看 success/fail再查前后条数差两个都对上这次压测结果才可信。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 报错怎么定位压测链路里报错分两类数据库侧的报错和模型调用侧的报错。数据库侧的报错相对直观模型调用侧的报错因为多了一层网络和鉴权排查起来容易绕弯路。这一节把常见的几类报错和定位方法列出来。第一类数据库连接报错pymysql.err.OperationalError: (2003, Cant connect to MySQL server on 127.0.0.1)这种是连不上。先确认端口对不对Mysql 是 3306Doris 是 9030。再确认服务有没有起telnet 127.0.0.1 9030试一下。如果 Doris 的 FE 起了但 BE 没起连接能建立但写入会失败报错通常是Failed to find backend。pymysql.err.OperationalError: (1040, Too many connections)是连接数打满。把maxconnections调小或者把blocking设为 True 让线程排队。Doris 侧尤其要注意FE 的连接数上限默认不高压测线程别开太多。pymysql.err.IntegrityError: (1062, Duplicate entry xxx for key PRIMARY)是主键冲突。检查 data_id 生成逻辑确保多线程之间不重叠。原始 excerpt 里thread_id * count i 1这个公式如果count和实际批次不一致就会撞。第二类模型调用报错401 Unauthorized是最常见的。先检查.env里的TAOTOKEN_API_KEY有没有写错有没有多余空格。然后确认 Base URL 是不是https://taotoken.net/api注意结尾不要多加/v1SDK 会自己拼。如果用的是 Anthropic SDKBase URL 的拼法可能不同参考接入文档里的说明。local proxy failed这种报错通常和本地网络环境有关。先确认你的请求是直接发到 TaoToken 的 API 入口没有经过额外的本地转发。检查.env里有没有残留的HTTP_PROXY或HTTPS_PROXY环境变量有的话清掉。Python 里可以用os.environ.pop(HTTP_PROXY, None)在脚本开头清一下。Error reading choices或reading choices这类报错通常是响应体解析失败。原因可能是模型返回了非预期格式或者请求被截断。先确认TAOTOKEN_MODEL_ID填的模型名是存在的去模型对话页面确认一下。然后把temperature调低max_tokens设大一点避免返回被截断。OAuth相关报错如果你用的是 Claude Code 这类工具可能是工具的认证方式和 API Key 方式冲突了。Claude Code 的配置入口在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite 按里面的说明配 Base URL 和 Key。如果工具里同时开了 OAuth 和 API Key可能会互相覆盖建议只保留一种。第三类压测结果异常压测跑完 QPS 特别低先看total_time里有没有包含连接池等待时间。如果blockingTrue且maxconnections太小线程大部分时间在等连接QPS 自然上不去。把maxconnections调到和线程数匹配。P99 特别高但 P50 正常通常是偶发的连接重建或 Doris compaction。看压测期间有没有网络抖动或者 Doris 是不是在后台做 compaction。这种情况多跑几次取稳定值。total_fail大于 0 但数据库条数对得上可能是失败后重试成功了。检查脚本里有没有自动重试逻辑如果有fail 计数和实际失败数会不一致。建议失败就记录不自动重试压测结束后统一对账。第四类配置类报错KeyError: TAOTOKEN_API_KEY是.env没加载。确认python-dotenv装了load_dotenv()在读取环境变量之前调用了。如果.env文件和脚本不在同一目录load_dotenv()要传路径。ModuleNotFoundError: No module named dbutils是依赖没装。pip install dbutils pymysql python-dotenv openai一把装齐。Doris 侧如果报Unknown table tbl_point_query1先确认表建了没有库名对不对。Doris 的库表是大小写敏感的test.tbl_point_query1和test.TBL_POINT_QUERY1是两张表。排查的核心思路是先分清是数据库侧还是模型侧再看是连接问题还是数据问题最后看是配置问题还是代码问题。每一类报错都有对应的检查点按顺序过一遍大部分问题都能定位到。6. 把压测、校验、分析串成一条可复用的链路到这里Mysql 和 Doris 的压测脚本、连接池配置、结果校验、模型调用都已经能跑通了。最后说一下怎么把这些串成一条可复用的链路而不是每次压测都手动拼。我的做法是建一个bench_runner.py把压测、校验、分析三步串起来# bench_runner.py from mysql_bench import run_mysql_bench from doris_bench import run_doris_bench from verify import count_rows, verify_bench from llm_helper import analyze_result from config import MYSQL_CONFIG, DORIS_CONFIG def run_full_bench(mysql_threads50, mysql_batch20, doris_threads40, doris_batch25): before_mysql count_rows(MYSQL_CONFIG) before_doris count_rows(DORIS_CONFIG) mysql_result run_mysql_bench(mysql_threads, mysql_batch) doris_result run_doris_bench(doris_threads, doris_batch) verify_bench(before_mysql, before_doris, mysql_result, doris_result) analysis analyze_result(mysql_result, doris_result) print(模型分析结果) print(analysis) return {mysql: mysql_result, doris: doris_result, analysis: analysis} if __name__ __main__: run_full_bench()这样每次压测只需要调run_full_bench()传入线程数和批次剩下的校验和分析自动完成。模型分析那一步走的是 TaoToken 的统一 Key和压测脚本共用.env里的配置。如果你要长期做压测建议把每次结果存下来方便对比。可以存成 JSON字段包括时间戳、线程数、批次、QPS、P50、P99、success、fail。下次压测时把历史结果一起丢给模型让模型帮你判断这次是变好了还是变差了。Coding Plan 的入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 如果你要长期跑压测和 Agent 类任务可以看一下那边的方案适合需要持续调用模型的场景。最后说一个我踩过的坑Doris 压测时不要用太高的并发去压单 BE 节点BE 的写入线程池有限并发过高会排队QPS 反而下降。我试过把线程从 40 加到 80QPS 没涨P99 翻了一倍。后来降到 40QPS 反而更稳。所以压测参数不是越大越好找到拐点比堆线程更重要。整套链路的核心就三件事压测脚本要能区分 Mysql 和 Doris 的写入语义结果校验要以数据库实际条数为准模型调用要统一 Key 避免到处配。这三件做好压测结果才敢拿去用。