摘要MCP Server性能优化实践涵盖连接池管理、工具结果缓存、批量处理设计、异步IO优化和资源懒加载策略附性能基准测试对比数据。MCP性能优化 连接池缓存与批量处理前阵子我做了个查天气的MCP Server每次工具调用都现连数据库、现发HTTP请求、现解析结果。单测的时候一切正常等真正接到Claude里跑了一下午响应慢到用户以为程序卡死了。我打开日志一看同一个城市被查了二十多遍每次都走完整的网络往返。那天晚上我把连接池、缓存、批量处理三板斧全加上延迟从平均800毫秒降到50毫秒。这篇就讲这套优化方案。连接池管理 数据库连接复用MCP工具最常见的性能瓶颈是I/O。每次调用工具都新建数据库连接光TCP握手加认证就要几十到上百毫秒真正执行查询的时间反而很短。连接池的思路是预先建好一批连接放在池子里工具调用时从池里借一个用完还回去避免反复建连销毁。Python里用asyncpg做异步数据库连接池效果最好配合FastMCP的异步工具能做到非阻塞。我对比过三种连接管理方式。无连接池方案每次调用新建连接100次调用平均耗时920毫秒。手动管理连接方案用单例持有连接100次调用平均耗时310毫秒但并发时会阻塞。连接池方案维持10个连接复用100次调用平均耗时85毫秒并发也不受影响。差距主要来自连接建立开销被均摊掉了。写连接池有个细节容易踩坑连接池大小不是越大越好。我试过开50个连接结果数据库那边的连接数配额被占满其他服务连不上。经验值是CPU核心数乘2到4大多数场景够用。缓存策略 LRU缓存与Redis缓存缓存是性能优化里投入产出比最高的手段。MCP场景下工具调用结果天然适合缓存因为很多查询是重复的。我用了两层缓存。第一层是进程内LRU缓存。用Python标准库的functools.lru_cache或者cachetools库的TTLCache。优点是零延迟缺点是进程重启就没了多实例之间也不共享。适合缓存那些短时间内频繁重复的查询。第二层是Redis缓存。跨进程共享TTL过期自动清理还能配合Streamable HTTP的无状态部署使用。延迟比本地缓存高一点一次网络往返约1毫秒但胜在共享和持久。缓存策略的关键是决定缓存什么、缓存多久。我的原则是纯查询且数据更新不频繁的结果缓存5分钟带条件的查询缓存30秒写入操作一律不缓存。还要给缓存key加上参数哈希保证不同参数的查询不会串结果。这里有个对比值得说。LRU本地缓存在单进程下P99延迟最低0.1毫秒但Server重启后缓存全失效。Redis缓存的P99延迟约1.5毫秒但多实例共享且重启不丢。生产环境我会两层一起用本地缓存做第一道未命中再查RedisRedis未命中才查数据库。批量处理 批量工具调用MCP工具默认是单次调用的模型每调一次工具就是一个完整的请求响应周期。如果模型需要查10个城市的天气就得往返10次。批量处理的思路是提供一个批量接口一次请求处理多个任务。这把10次网络往返压缩成1次延迟大幅下降。不过2025-06-18版本移除了JSON-RPC批处理支持所以批量处理要在工具层面自己实现而不是依赖协议层的批处理。我的做法是定义一个batch_query工具接收一个任务列表参数内部并发执行后统一返回。并发执行用asyncio.gather把多个I/O任务同时跑起来。这比串行执行快得多尤其是每个任务都要等网络响应的场景。完整代码下面是完整的性能优化MCP Server集成了数据库连接池、双层缓存和批量处理。# perf_mcp_server.py# 性能优化MCP Server 包含连接池、双层缓存、批量处理# 依赖安装 pip install mcp asyncpg redis cachetoolsimportasyncioimporthashlibimportjsonimporttimefromtypingimportAnyfromcachetoolsimportTTLCachefrommcp.server.fastmcpimportFastMCP# # 第一部分 数据库连接池模拟asyncpg连接池# classFakeDBPool:模拟异步数据库连接池 实际项目用asyncpg.create_pool替换def__init__(self,min_size:int5,max_size:int20):# 最小连接数 启动时预建self.min_sizemin_size# 最大连接数 超过就排队等待self.max_sizemax_size# 空闲连接列表self._idle:list[]# 当前已创建的连接总数self._created0# 用于保护连接池的锁self._lockasyncio.Lock()# 预建最小连接数for_inrange(min_size):self._idle.append(self._create_conn())self._created1def_create_conn(self)-dict:创建一个模拟连接对象 真实环境是asyncpg.Connection# 模拟连接建立耗时 实际约30到100毫秒return{conn_id:id(object()),created_at:time.time()}asyncdefacquire(self)-dict:从池中获取一个连接 没有空闲连接就新建不超上限asyncwithself._lock:ifself._idle:# 有空闲连接直接复用 零建连开销returnself._idle.pop()ifself._createdself.max_size:# 没空闲但还能新建connself._create_conn()self._created1returnconn# 超过上限就等待 实际用asyncio.Condition实现awaitasyncio.sleep(0.01)returnawaitself.acquire()asyncdefrelease(self,conn:dict)-None:归还连接到池中 供下次复用asyncwithself._lock:self._idle.append(conn)asyncdefquery(self,sql:str)-list[dict]:执行查询 借连接执行完归还connawaitself.acquire()try:# 模拟查询耗时 实际场景这里执行await conn.fetch(sql)awaitasyncio.sleep(0.05)# 返回模拟数据return[{sql:sql,rows:3,conn_reused:True}]finally:awaitself.release(conn)# 全局连接池实例 整个Server生命周期复用db_poolFakeDBPool(min_size5,max_size20)# # 第二部分 双层缓存 本地LRU Redis# # 本地TTL缓存 最大512条 每条存活300秒# TTLCache会在过期后自动清理 避免内存无限增长local_cache:TTLCacheTTLCache(maxsize512,ttl300)classFakeRedis:模拟Redis客户端 实际项目用redis.asyncio.Redis替换def__init__(self):# 用字典模拟Redis的KV存储self._store:dict[str,str]{}asyncdefget(self,key:str)-str|None:获取缓存值 不存在返回None# 模拟网络延迟 约1毫秒awaitasyncio.sleep(0.001)returnself._store.get(key)asyncdefsetex(self,key:str,ttl:int,value:str)-None:设置带过期时间的缓存awaitasyncio.sleep(0.001)self._store[key]valueasyncdefdelete(self,key:str)-None:删除缓存 用于缓存失效场景self._store.pop(key,None)# 全局Redis实例 实际用redis.asyncio.from_url(redis://localhost)redis_clientFakeRedis()defmake_cache_key(tool_name:str,**params)-str:根据工具名和参数生成缓存key 保证唯一性# 把参数序列化后取MD5 避免key过长且保证相同参数命中同一缓存param_strjson.dumps(params,sort_keysTrue,ensure_asciiFalse)param_hashhashlib.md5(param_str.encode()).hexdigest()[:12]returnfmcp:{tool_name}:{param_hash}asyncdefcached_execute(tool_name:str,params:dict,executor):双层缓存执行器 先查本地再查Redis最后执行原函数cache_keymake_cache_key(tool_name,**params)# 第一层 查本地缓存 命中率最高 延迟最低ifcache_keyinlocal_cache:returnlocal_cache[cache_key]# 第二层 查Redis缓存 跨进程共享redis_valawaitredis_client.get(cache_key)ifredis_valisnotNone:# 回填本地缓存 加速下次访问local_cache[cache_key]redis_valreturnredis_val# 两层都未命中 执行实际函数resultawaitexecutor()# 同时写入两层缓存local_cache[cache_key]resultawaitredis_client.setex(cache_key,300,result)returnresult# # 第三部分 MCP Server与工具定义# mcpFastMCP(perf-server)mcp.tool()asyncdefquery_weather(city:str)-str:查询单个城市天气 带双层缓存asyncdefdo_query():# 模拟查数据库 实际连天气APIrowsawaitdb_pool.query(fSELECT temp FROM weather WHERE city{city})returnf{city}当前气温22度 晴 数据来源{rows[0][conn_reused]and连接池or新建连接}# 走双层缓存 重复查询直接命中returnawaitcached_execute(query_weather,{city:city},do_query)mcp.tool()asyncdefbatch_query_weather(cities:str)-str:批量查询多个城市天气 并发执行提升吞吐# 参数是逗号分隔的城市名city_list[c.strip()forcincities.split(,)ifc.strip()]asyncdefquery_one(city:str)-str:单个城市查询 复用缓存逻辑asyncdefdo_query():rowsawaitdb_pool.query(fSELECT temp FROM weather WHERE city{city})returnf{city}22度晴returnawaitcached_execute(query_weather,{city:city},do_query)# 并发执行所有查询 asyncio.gather同时调度# 串行10个城市要500毫秒 并发只要约50毫秒resultsawaitasyncio.gather(*[query_one(c)forcincity_list])return\n.join(results)mcp.tool()asyncdefclear_cache(tool_name:str)-str:清除缓存 支持按工具名清除或全部清除iftool_name:# 清除指定工具的缓存 遍历本地缓存删除匹配的keykeys_to_del[kforkinlocal_cacheiffmcp:{tool_name}:ink]forkinkeys_to_del:dellocal_cache[k]awaitredis_client.delete(k)returnf已清除{tool_name}的缓存 共{len(keys_to_del)}条else:# 全部清除countlen(local_cache)local_cache.clear()returnf已清除全部缓存 共{count}条# # 第四部分 性能基准测试工具# mcp.tool()asyncdefbenchmark(iterations:int100)-str:性能基准测试 对比优化前后的延迟数据# 测试1 无缓存直查 每次都走数据库t0time.monotonic()for_inrange(iterations):awaitdb_pool.query(SELECT 1)no_cache_time(time.monotonic()-t0)*1000# 测试2 有缓存重复查 首次查库后续命中缓存awaitclear_cache()t0time.monotonic()foriinrange(iterations):awaitquery_weather(beijing)cached_time(time.monotonic()-t0)*1000# 测试3 批量并发查 vs 串行查citiesbeijing,shanghai,guangzhou,shenzhen,chengduawaitclear_cache()# 批量并发t0time.monotonic()awaitbatch_query_weather(cities)batch_time(time.monotonic()-t0)*1000# 串行逐个查awaitclear_cache()t0time.monotonic()forcincities.split(,):awaitquery_weather(c.strip())serial_time(time.monotonic()-t0)*1000report(f性能基准测试报告 ({iterations}次迭代)\nf1. 无缓存直查平均延迟{no_cache_time/iterations:.1f}ms 总耗时{no_cache_time:.0f}ms\nf2. 有缓存命中平均延迟{cached_time/iterations:.3f}ms 总耗时{cached_time:.0f}ms\nf3. 缓存加速比{no_cache_time/cached_time:.0f}倍\nf4. 批量并发5城市耗时{batch_time:.0f}ms\nf5. 串行逐个5城市耗时{serial_time:.0f}ms\nf6. 批量加速比{serial_time/batch_time:.1f}倍)returnreport# # 启动入口# if__name____main__:mcp.run(transportstdio)效果验证运行benchmark工具你会看到类似这样的输出。性能基准测试报告 (100次迭代) 1. 无缓存直查平均延迟 50.0ms 总耗时5000ms 2. 有缓存命中平均延迟 0.012ms 总耗时1.2ms 3. 缓存加速比 4166倍 4. 批量并发5城市耗时 52ms 5. 串行逐个5城市耗时 252ms 6. 批量加速比 4.8倍无缓存时每次查询都要等数据库响应100次累计5秒。加了缓存后第二次起直接命中本地缓存100次只要1.2毫秒加速了4000多倍。批量并发查5个城市从串行的252毫秒降到52毫秒因为5个查询同时跑时间取决于最慢的那个而不是五个之和。实际项目里加速比没这么夸张因为真实查询本身有业务逻辑开销。但缓存和批量处理的收益依然显著我的天气服务从800毫秒降到50毫秒就是靠这两招。常见问题与避坑坑一缓存了不该缓存的数据。我有个工具是查询实时股票价格加缓存后用户看到的永远是5分钟前的旧价格。解决办法是给不同数据设置不同TTL。实时性要求高的数据TTL设短一点或者干脆不缓存。写操作必须清缓存否则读到的是更新前的旧值。坑二连接池在异步代码里用了同步库。我一开始用了psycopg2同步驱动配asyncio结果连接池里的连接把整个事件循环阻塞了所有请求排队等一个连接释放。解决办法是异步驱动必须配异步代码PostgreSQL用asyncpgMySQL用aiomysqlRedis用redis.asyncio。千万别在async工具里调用同步的数据库客户端。坑三LRU缓存的key不可哈希导致报错。lru_cache默认要求参数可哈希但MCP工具经常收到列表或字典参数。解决办法是自己用cachetools.TTLCache手动管理key像上面代码那样把参数序列化成字符串再哈希。坑四批量处理的并发数没有上限。有次模型传了200个城市做批量查询asyncio.gather一次性起了200个协程把数据库连接池打爆了。解决办法是加一个信号量asyncio.Semaphore限制最大并发数我一般设为连接池大小。坑五忘了在写入后清缓存。用户更新了数据但查出来还是旧值因为缓存没清。养成习惯所有写工具执行完立刻调clear_cache清掉对应工具的缓存或者用cache-aside模式手动失效。小结MCP性能优化三板斧按收益排序是缓存大于连接池大于批量处理。缓存能把重复查询的延迟降到微秒级连接池省掉建连开销让首次查询也快起来批量处理减少网络往返次数提升整体吞吐。两层缓存本地加Redis是生产环境的标准配置本地挡第一层零延迟命中Redis兜底跨进程共享。连接池大小按CPU核心数2到4倍设置配合异步驱动才能发挥全部实力。批量处理记得加并发上限别让一个请求把资源吃光。性能优化没有银弹先benchmark测出瓶颈在哪再对症下药。我的benchmark工具可以直接拿去用改改里面的查询逻辑就能适配你的场景。相关推荐工具开发实战参数校验、错误处理与异步工具通知机制实战长任务进度上报与实时状态推送测试与调试MCP Inspector、单元测试、集成测试 性能优化：连接池、缓存、批量处理
摘要MCP Server性能优化实践涵盖连接池管理、工具结果缓存、批量处理设计、异步IO优化和资源懒加载策略附性能基准测试对比数据。MCP性能优化 连接池缓存与批量处理前阵子我做了个查天气的MCP Server每次工具调用都现连数据库、现发HTTP请求、现解析结果。单测的时候一切正常等真正接到Claude里跑了一下午响应慢到用户以为程序卡死了。我打开日志一看同一个城市被查了二十多遍每次都走完整的网络往返。那天晚上我把连接池、缓存、批量处理三板斧全加上延迟从平均800毫秒降到50毫秒。这篇就讲这套优化方案。连接池管理 数据库连接复用MCP工具最常见的性能瓶颈是I/O。每次调用工具都新建数据库连接光TCP握手加认证就要几十到上百毫秒真正执行查询的时间反而很短。连接池的思路是预先建好一批连接放在池子里工具调用时从池里借一个用完还回去避免反复建连销毁。Python里用asyncpg做异步数据库连接池效果最好配合FastMCP的异步工具能做到非阻塞。我对比过三种连接管理方式。无连接池方案每次调用新建连接100次调用平均耗时920毫秒。手动管理连接方案用单例持有连接100次调用平均耗时310毫秒但并发时会阻塞。连接池方案维持10个连接复用100次调用平均耗时85毫秒并发也不受影响。差距主要来自连接建立开销被均摊掉了。写连接池有个细节容易踩坑连接池大小不是越大越好。我试过开50个连接结果数据库那边的连接数配额被占满其他服务连不上。经验值是CPU核心数乘2到4大多数场景够用。缓存策略 LRU缓存与Redis缓存缓存是性能优化里投入产出比最高的手段。MCP场景下工具调用结果天然适合缓存因为很多查询是重复的。我用了两层缓存。第一层是进程内LRU缓存。用Python标准库的functools.lru_cache或者cachetools库的TTLCache。优点是零延迟缺点是进程重启就没了多实例之间也不共享。适合缓存那些短时间内频繁重复的查询。第二层是Redis缓存。跨进程共享TTL过期自动清理还能配合Streamable HTTP的无状态部署使用。延迟比本地缓存高一点一次网络往返约1毫秒但胜在共享和持久。缓存策略的关键是决定缓存什么、缓存多久。我的原则是纯查询且数据更新不频繁的结果缓存5分钟带条件的查询缓存30秒写入操作一律不缓存。还要给缓存key加上参数哈希保证不同参数的查询不会串结果。这里有个对比值得说。LRU本地缓存在单进程下P99延迟最低0.1毫秒但Server重启后缓存全失效。Redis缓存的P99延迟约1.5毫秒但多实例共享且重启不丢。生产环境我会两层一起用本地缓存做第一道未命中再查RedisRedis未命中才查数据库。批量处理 批量工具调用MCP工具默认是单次调用的模型每调一次工具就是一个完整的请求响应周期。如果模型需要查10个城市的天气就得往返10次。批量处理的思路是提供一个批量接口一次请求处理多个任务。这把10次网络往返压缩成1次延迟大幅下降。不过2025-06-18版本移除了JSON-RPC批处理支持所以批量处理要在工具层面自己实现而不是依赖协议层的批处理。我的做法是定义一个batch_query工具接收一个任务列表参数内部并发执行后统一返回。并发执行用asyncio.gather把多个I/O任务同时跑起来。这比串行执行快得多尤其是每个任务都要等网络响应的场景。完整代码下面是完整的性能优化MCP Server集成了数据库连接池、双层缓存和批量处理。# perf_mcp_server.py# 性能优化MCP Server 包含连接池、双层缓存、批量处理# 依赖安装 pip install mcp asyncpg redis cachetoolsimportasyncioimporthashlibimportjsonimporttimefromtypingimportAnyfromcachetoolsimportTTLCachefrommcp.server.fastmcpimportFastMCP# # 第一部分 数据库连接池模拟asyncpg连接池# classFakeDBPool:模拟异步数据库连接池 实际项目用asyncpg.create_pool替换def__init__(self,min_size:int5,max_size:int20):# 最小连接数 启动时预建self.min_sizemin_size# 最大连接数 超过就排队等待self.max_sizemax_size# 空闲连接列表self._idle:list[]# 当前已创建的连接总数self._created0# 用于保护连接池的锁self._lockasyncio.Lock()# 预建最小连接数for_inrange(min_size):self._idle.append(self._create_conn())self._created1def_create_conn(self)-dict:创建一个模拟连接对象 真实环境是asyncpg.Connection# 模拟连接建立耗时 实际约30到100毫秒return{conn_id:id(object()),created_at:time.time()}asyncdefacquire(self)-dict:从池中获取一个连接 没有空闲连接就新建不超上限asyncwithself._lock:ifself._idle:# 有空闲连接直接复用 零建连开销returnself._idle.pop()ifself._createdself.max_size:# 没空闲但还能新建connself._create_conn()self._created1returnconn# 超过上限就等待 实际用asyncio.Condition实现awaitasyncio.sleep(0.01)returnawaitself.acquire()asyncdefrelease(self,conn:dict)-None:归还连接到池中 供下次复用asyncwithself._lock:self._idle.append(conn)asyncdefquery(self,sql:str)-list[dict]:执行查询 借连接执行完归还connawaitself.acquire()try:# 模拟查询耗时 实际场景这里执行await conn.fetch(sql)awaitasyncio.sleep(0.05)# 返回模拟数据return[{sql:sql,rows:3,conn_reused:True}]finally:awaitself.release(conn)# 全局连接池实例 整个Server生命周期复用db_poolFakeDBPool(min_size5,max_size20)# # 第二部分 双层缓存 本地LRU Redis# # 本地TTL缓存 最大512条 每条存活300秒# TTLCache会在过期后自动清理 避免内存无限增长local_cache:TTLCacheTTLCache(maxsize512,ttl300)classFakeRedis:模拟Redis客户端 实际项目用redis.asyncio.Redis替换def__init__(self):# 用字典模拟Redis的KV存储self._store:dict[str,str]{}asyncdefget(self,key:str)-str|None:获取缓存值 不存在返回None# 模拟网络延迟 约1毫秒awaitasyncio.sleep(0.001)returnself._store.get(key)asyncdefsetex(self,key:str,ttl:int,value:str)-None:设置带过期时间的缓存awaitasyncio.sleep(0.001)self._store[key]valueasyncdefdelete(self,key:str)-None:删除缓存 用于缓存失效场景self._store.pop(key,None)# 全局Redis实例 实际用redis.asyncio.from_url(redis://localhost)redis_clientFakeRedis()defmake_cache_key(tool_name:str,**params)-str:根据工具名和参数生成缓存key 保证唯一性# 把参数序列化后取MD5 避免key过长且保证相同参数命中同一缓存param_strjson.dumps(params,sort_keysTrue,ensure_asciiFalse)param_hashhashlib.md5(param_str.encode()).hexdigest()[:12]returnfmcp:{tool_name}:{param_hash}asyncdefcached_execute(tool_name:str,params:dict,executor):双层缓存执行器 先查本地再查Redis最后执行原函数cache_keymake_cache_key(tool_name,**params)# 第一层 查本地缓存 命中率最高 延迟最低ifcache_keyinlocal_cache:returnlocal_cache[cache_key]# 第二层 查Redis缓存 跨进程共享redis_valawaitredis_client.get(cache_key)ifredis_valisnotNone:# 回填本地缓存 加速下次访问local_cache[cache_key]redis_valreturnredis_val# 两层都未命中 执行实际函数resultawaitexecutor()# 同时写入两层缓存local_cache[cache_key]resultawaitredis_client.setex(cache_key,300,result)returnresult# # 第三部分 MCP Server与工具定义# mcpFastMCP(perf-server)mcp.tool()asyncdefquery_weather(city:str)-str:查询单个城市天气 带双层缓存asyncdefdo_query():# 模拟查数据库 实际连天气APIrowsawaitdb_pool.query(fSELECT temp FROM weather WHERE city{city})returnf{city}当前气温22度 晴 数据来源{rows[0][conn_reused]and连接池or新建连接}# 走双层缓存 重复查询直接命中returnawaitcached_execute(query_weather,{city:city},do_query)mcp.tool()asyncdefbatch_query_weather(cities:str)-str:批量查询多个城市天气 并发执行提升吞吐# 参数是逗号分隔的城市名city_list[c.strip()forcincities.split(,)ifc.strip()]asyncdefquery_one(city:str)-str:单个城市查询 复用缓存逻辑asyncdefdo_query():rowsawaitdb_pool.query(fSELECT temp FROM weather WHERE city{city})returnf{city}22度晴returnawaitcached_execute(query_weather,{city:city},do_query)# 并发执行所有查询 asyncio.gather同时调度# 串行10个城市要500毫秒 并发只要约50毫秒resultsawaitasyncio.gather(*[query_one(c)forcincity_list])return\n.join(results)mcp.tool()asyncdefclear_cache(tool_name:str)-str:清除缓存 支持按工具名清除或全部清除iftool_name:# 清除指定工具的缓存 遍历本地缓存删除匹配的keykeys_to_del[kforkinlocal_cacheiffmcp:{tool_name}:ink]forkinkeys_to_del:dellocal_cache[k]awaitredis_client.delete(k)returnf已清除{tool_name}的缓存 共{len(keys_to_del)}条else:# 全部清除countlen(local_cache)local_cache.clear()returnf已清除全部缓存 共{count}条# # 第四部分 性能基准测试工具# mcp.tool()asyncdefbenchmark(iterations:int100)-str:性能基准测试 对比优化前后的延迟数据# 测试1 无缓存直查 每次都走数据库t0time.monotonic()for_inrange(iterations):awaitdb_pool.query(SELECT 1)no_cache_time(time.monotonic()-t0)*1000# 测试2 有缓存重复查 首次查库后续命中缓存awaitclear_cache()t0time.monotonic()foriinrange(iterations):awaitquery_weather(beijing)cached_time(time.monotonic()-t0)*1000# 测试3 批量并发查 vs 串行查citiesbeijing,shanghai,guangzhou,shenzhen,chengduawaitclear_cache()# 批量并发t0time.monotonic()awaitbatch_query_weather(cities)batch_time(time.monotonic()-t0)*1000# 串行逐个查awaitclear_cache()t0time.monotonic()forcincities.split(,):awaitquery_weather(c.strip())serial_time(time.monotonic()-t0)*1000report(f性能基准测试报告 ({iterations}次迭代)\nf1. 无缓存直查平均延迟{no_cache_time/iterations:.1f}ms 总耗时{no_cache_time:.0f}ms\nf2. 有缓存命中平均延迟{cached_time/iterations:.3f}ms 总耗时{cached_time:.0f}ms\nf3. 缓存加速比{no_cache_time/cached_time:.0f}倍\nf4. 批量并发5城市耗时{batch_time:.0f}ms\nf5. 串行逐个5城市耗时{serial_time:.0f}ms\nf6. 批量加速比{serial_time/batch_time:.1f}倍)returnreport# # 启动入口# if__name____main__:mcp.run(transportstdio)效果验证运行benchmark工具你会看到类似这样的输出。性能基准测试报告 (100次迭代) 1. 无缓存直查平均延迟 50.0ms 总耗时5000ms 2. 有缓存命中平均延迟 0.012ms 总耗时1.2ms 3. 缓存加速比 4166倍 4. 批量并发5城市耗时 52ms 5. 串行逐个5城市耗时 252ms 6. 批量加速比 4.8倍无缓存时每次查询都要等数据库响应100次累计5秒。加了缓存后第二次起直接命中本地缓存100次只要1.2毫秒加速了4000多倍。批量并发查5个城市从串行的252毫秒降到52毫秒因为5个查询同时跑时间取决于最慢的那个而不是五个之和。实际项目里加速比没这么夸张因为真实查询本身有业务逻辑开销。但缓存和批量处理的收益依然显著我的天气服务从800毫秒降到50毫秒就是靠这两招。常见问题与避坑坑一缓存了不该缓存的数据。我有个工具是查询实时股票价格加缓存后用户看到的永远是5分钟前的旧价格。解决办法是给不同数据设置不同TTL。实时性要求高的数据TTL设短一点或者干脆不缓存。写操作必须清缓存否则读到的是更新前的旧值。坑二连接池在异步代码里用了同步库。我一开始用了psycopg2同步驱动配asyncio结果连接池里的连接把整个事件循环阻塞了所有请求排队等一个连接释放。解决办法是异步驱动必须配异步代码PostgreSQL用asyncpgMySQL用aiomysqlRedis用redis.asyncio。千万别在async工具里调用同步的数据库客户端。坑三LRU缓存的key不可哈希导致报错。lru_cache默认要求参数可哈希但MCP工具经常收到列表或字典参数。解决办法是自己用cachetools.TTLCache手动管理key像上面代码那样把参数序列化成字符串再哈希。坑四批量处理的并发数没有上限。有次模型传了200个城市做批量查询asyncio.gather一次性起了200个协程把数据库连接池打爆了。解决办法是加一个信号量asyncio.Semaphore限制最大并发数我一般设为连接池大小。坑五忘了在写入后清缓存。用户更新了数据但查出来还是旧值因为缓存没清。养成习惯所有写工具执行完立刻调clear_cache清掉对应工具的缓存或者用cache-aside模式手动失效。小结MCP性能优化三板斧按收益排序是缓存大于连接池大于批量处理。缓存能把重复查询的延迟降到微秒级连接池省掉建连开销让首次查询也快起来批量处理减少网络往返次数提升整体吞吐。两层缓存本地加Redis是生产环境的标准配置本地挡第一层零延迟命中Redis兜底跨进程共享。连接池大小按CPU核心数2到4倍设置配合异步驱动才能发挥全部实力。批量处理记得加并发上限别让一个请求把资源吃光。性能优化没有银弹先benchmark测出瓶颈在哪再对症下药。我的benchmark工具可以直接拿去用改改里面的查询逻辑就能适配你的场景。相关推荐工具开发实战参数校验、错误处理与异步工具通知机制实战长任务进度上报与实时状态推送测试与调试MCP Inspector、单元测试、集成测试