1. 高并发爬取后写库为什么会卡死Scrapy 默认是单线程事件循环驱动的异步框架但很多人在管道里写数据库时用的是同步的pymysql.connect()或MySQLdb.connect()每次process_item都新建连接、插入、关闭。爬取速度一上来问题立刻暴露连接数暴涨、MySQL 报Too many connections、事件循环被阻塞导致下载队列堆积、内存飙升。这个场景的核心矛盾是Scrapy 的异步调度和数据库的同步阻塞调用不匹配。你爬得越快写库越慢最后整个爬虫被写库拖死。解决办法不是加机器而是把写库也变成异步的——用 Twisted 的adbapi.ConnectionPool把数据库操作丢进线程池主事件循环继续跑连接复用而不是每次新建。这篇聚焦工程落地在settings.json里配置 TaoToken 统一 Key用于爬虫里调用的模型接口比如智能解析、字段补全在管道里用连接池做批量异步插入最后给出可复制的压测验证动作。适合已经写过基础 Scrapy 管道、想解决写库瓶颈的开发者。TaoToken 在这里的角色是当你的爬虫需要在管道里调用大模型做内容清洗、字段抽取、分类打标时用统一 Key 管理所有模型请求不用在代码里散落多个平台的密钥。配置骨架和写库管道是两件独立的事但经常同时出现在一个项目里所以放在一起讲清楚。2. TaoToken 统一 Key 的前置准备先说清楚TaoToken 不是数据库也不是连接池本身。它是模型调用的统一入口。你的爬虫管道里如果有一步是「把抓到的文本丢给模型做结构化」那这一步需要 Key。统一 Key 的好处是一个 Key 走所有模型settings.json里只维护一处换模型不用改代码逻辑。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 根地址是 https://taotoken.net/api 这个不加 UTM。你需要先去控制台创建 API Key控制台入口在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。拿到 Key 之后不要硬编码在pipelines.py里。正确做法是写进settings.json通过环境变量或配置文件读取。下面给一个settings.json的骨架把数据库配置和 TaoToken 配置分开管理{ mysql: { host: 127.0.0.1, port: 3306, user: root, password: your_db_password, dbname: spider_db, charset: utf8mb4 }, taotoken: { api_base: https://taotoken.net/api, api_key: sk-你的统一Key, model: claude-3-5-sonnet, timeout: 30 }, pool: { min: 3, max: 20, idle_timeout: 300 } }然后在settings.py里加载这个 JSON暴露成 Scrapy 能读的配置项import json import os with open(os.path.join(os.path.dirname(__file__), settings.json), r, encodingutf-8) as f: _cfg json.load(f) MYSQL_HOST _cfg[mysql][host] MYSQL_PORT _cfg[mysql][port] MYSQL_USER _cfg[mysql][user] MYSQL_PASSWORD _cfg[mysql][password] MYSQL_DBNAME _cfg[mysql][dbname] MYSQL_CHARSET _cfg[mysql][charset] TAOTOKEN_API_BASE _cfg[taotoken][api_base] TAOTOKEN_API_KEY _cfg[taotoken][api_key] TAOTOKEN_MODEL _cfg[taotoken][model] POOL_MIN _cfg[pool][min] POOL_MAX _cfg[pool][max] ITEM_PIPELINES { projectname.pipelines.MySQLTwistedPipeline: 300, }注意settings.json不要提交到公开仓库Key 用环境变量覆盖更安全。可以在settings.py里加一层os.environ.get(TAOTOKEN_API_KEY, _cfg[taotoken][api_key])。如果你只是做写库不需要模型调用那 TaoToken 这部分可以跳过但连接池配置照常。如果你的管道里有模型调用步骤统一 Key 能省掉多平台密钥管理的麻烦。模型对话调试可以用 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 先验证请求格式。3. 连接池参数与异步插入管道配置核心是twisted.enterprise.adbapi.ConnectionPool。它内部维护一个线程池每个线程持有一个数据库连接runInteraction把操作丢进线程执行返回 Deferred主事件循环不阻塞。先看连接池参数怎么定。min是最小连接数max是最大连接数。高并发爬取时max不要超过 MySQL 的max_connections减去其他服务的占用。一般单机爬虫max设 10 到 20 够用设太大反而增加数据库压力。idle_timeout控制空闲连接回收避免连接长时间挂着被数据库端断开。下面是完整的管道代码包含连接池初始化、异步插入、批量提交和异常处理import json import logging from twisted.enterprise import adbapi import pymysql from pymysql import cursors logger logging.getLogger(__name__) class MySQLTwistedPipeline(object): def __init__(self, dbpool): self.dbpool dbpool self.batch [] self.batch_size 50 classmethod def from_settings(cls, settings): dbparms dict( hostsettings[MYSQL_HOST], portsettings[MYSQL_PORT], usersettings[MYSQL_USER], passwordsettings[MYSQL_PASSWORD], databasesettings[MYSQL_DBNAME], charsetsettings[MYSQL_CHARSET], cursorclasscursors.DictCursor, use_unicodeTrue, autocommitFalse, ) dbpool adbapi.ConnectionPool( pymysql, **dbparms, cp_minsettings.getint(POOL_MIN, 3), cp_maxsettings.getint(POOL_MAX, 20), cp_idlesettings.getint(POOL_IDLE, 300), ) return cls(dbpool) def process_item(self, item, spider): self.batch.append(dict(item)) if len(self.batch) self.batch_size: batch self.batch self.batch [] d self.dbpool.runInteraction(self.do_batch_insert, batch) d.addErrback(self.handle_error, batch) return item def close_spider(self, spider): if self.batch: d self.dbpool.runInteraction(self.do_batch_insert, self.batch) d.addErrback(self.handle_error, self.batch) self.batch [] self.dbpool.close() def do_batch_insert(self, cursor, batch): sql INSERT INTO items (title, url, content, category, created_at) VALUES (%s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE content VALUES(content), category VALUES(category) params [ (row.get(title), row.get(url), row.get(content), row.get(category)) for row in batch ] cursor.executemany(sql, params) def handle_error(self, failure, batch): logger.error(批量插入失败条数%d原因%s, len(batch), failure.getErrorMessage()))几个关键点。第一executemany比循环execute快很多50 条一批是实测比较稳的粒度太小网络往返多太大单次事务锁时间长。第二autocommitFalse配合runInteraction的自动提交机制Twisted 会在do_batch_insert正常返回后提交异常时回滚。第三close_spider里要 flush 剩余批次并关闭连接池否则最后不足一批的数据会丢。如果你用 PostgreSQL把pymysql换成psycopg2SQL 里的ON DUPLICATE KEY UPDATE换成ON CONFLICT ... DO UPDATE其余结构不变。注意cp_min和cp_max是 Twisted 连接池的参数名不是min/max。写错不会报错但连接池行为不符合预期。4. 验证请求与成功结果配置写完先别急着跑全量爬虫。用一个小脚本单独验证连接池和插入逻辑确认没问题再接入 Scrapy。第一步验证 TaoToken Key 是否可用如果你的管道有模型调用curl -X POST https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-你的统一Key \ -H Content-Type: application/json \ -d { model: claude-3-5-sonnet, messages: [{role: user, content: 返回JSON: {\ok\: true}}] }返回里有choices字段就说明 Key 正常。如果返回 401检查 Key 是否复制完整返回 404检查api_base是否写成了https://taotoken.net/api而不是带/v1的完整路径。第二步验证连接池插入。写一个独立的test_pool.pyfrom twisted.internet import reactor from twisted.enterprise import adbapi import pymysql from pymysql import cursors dbpool adbapi.ConnectionPool( pymysql, host127.0.0.1, port3306, userroot, passwordyour_db_password, databasespider_db, charsetutf8mb4, cursorclasscursors.DictCursor, cp_min3, cp_max10, ) def do_insert(cursor, n): sql INSERT INTO items (title, url) VALUES (%s, %s) cursor.executemany(sql, [(ftitle_{i}, fhttp://example.com/{i}) for i in range(n)]) def done(result): print(插入成功结果:, result) reactor.stop() def error(failure): print(插入失败:, failure.getErrorMessage()) reactor.stop() d dbpool.runInteraction(do_insert, 100) d.addCallbacks(done, error) reactor.run()跑python test_pool.py看到「插入成功」并且数据库里多了 100 条记录说明连接池工作正常。然后去 MySQL 里查连接数SHOW STATUS LIKE Threads_connected;正常应该在cp_min到cp_max之间不会随着插入次数线性增长。如果连接数一直涨说明连接池没复用检查是不是每次process_item都新建了ConnectionPool。第三步接入 Scrapy 跑一个小规模爬取观察日志里有没有批量插入失败。用scrapy crawl yourspider -s LOG_LEVELINFO跑看close_spider时是否正常 flush。5. 本篇常见错误排查报错一ModuleNotFoundError: No module named MySQLdbTwisted 的adbapi.ConnectionPool第一个参数是模块名。如果你用pymysql就写pymysql不要写MySQLdb。pymysql的 API 和MySQLdb兼容但模块名不同。装了pymysql还报MySQLdb找不到就是这里写错了。报错二RuntimeError: reactor is already running在 Scrapy 里不要手动调reactor.run()Scrapy 自己管理 reactor。连接池在from_settings里创建由 Scrapy 的生命周期驱动。如果你在管道里写了reactor.run()删掉。报错三Too many connectionscp_max设太大了或者多个爬虫进程共用同一个数据库但各自开了大连接池。把cp_max降到 10 以内或者用数据库中间件统一管理。另外检查close_spider里有没有调self.dbpool.close()没关的话进程退出后连接不会立刻释放。报错四数据插入成功但数据库里没有autocommitFalse时如果do_batch_insert里没有触发提交数据不会落库。runInteraction正常返回会自动提交但如果你在函数里捕获了异常没抛出Twisted 以为成功了实际事务可能没提交。不要在do_batch_insert里吞异常让它抛出去addErrback会处理。报错五TaoToken 请求超时settings.json里的timeout设太小或者模型响应慢。把超时调到 60 秒并且在管道里对模型调用做重试。注意模型调用和数据库插入是两个独立的异步操作不要串在一个runInteraction里否则模型慢会占着数据库连接。报错六ON DUPLICATE KEY UPDATE不生效检查表上有没有唯一索引。ON DUPLICATE KEY依赖唯一键或主键冲突才会触发更新。如果url字段没有唯一索引重复插入不会更新会直接插入新行。6. 长期跑爬虫的配置管理建议如果你的爬虫是长期运行的或者要跑多个站点建议把模型调用和写库拆成两个管道。模型调用管道负责用 TaoToken 统一 Key 做内容处理写库管道只负责插入。这样模型接口出问题不会阻塞写库写库出问题也不影响模型处理。长期编码和 Agent 场景可以用 Coding Plan 管理调用额度入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。API Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。Claude Code 相关配置参考 https://taotoken.net/claude-code?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite 。连接池参数没有万能值。我的经验是先设cp_min3, cp_max10跑一轮压测看Threads_connected的峰值和插入耗时再往上调。批量大小从 50 开始试如果单批插入超过 200ms就降到 30。数据库的max_allowed_packet也要检查批量插入的 SQL 长度不能超过这个值否则会报Packet too large。
