scrapy pipeline中应避免同步写入mongodb，需采用攒批+异步flush策略：在open_spider中初始化带maxpoolsize的mongoclient，process_item缓存item并用reactor.calllater定时或按量触发insert_many（ordered=false），close_spider中清空buffer并关闭连接。
Scrapy Pipeline 中直接用 pymongo 同步写入会拖慢爬虫速度
Scrapy 的 process_item 是同步执行的，如果在 pipeline 里直接调用 collection.insert_one() 或 collection.insert_many()，每次写入都得等 MongoDB 返回确认，网络 RTT + 服务端处理时间会卡住整个 item 流水线。实测在千级 QPS 场景下，吞吐可能掉 40% 以上。
真正可行的做法是攒批 + 异步触发：缓存一批 item，达到阈值或超时后统一提交。注意 scrapy 本身不提供协程调度，所以不能直接用 asyncio + motor —— 因为 process_item 不在 event loop 中运行，强行 await 会报 RuntimeError: no running event loop。
- 别在
process_item里写await collection.insert_many(...) - 用
threading.Timer或reactor.callLater做延迟 flush（推荐后者，与 Scrapy 的 Twisted 集成更稳） - 缓冲区建议设为 100–500 条，太小导致频繁提交，太大内存压力高且重启易丢数
MongoPipeline 必须重写 open_spider 和 close_spider 生命周期方法
连接 MongoDB 不能放在 __init__ 里——Scrapy 会为每个 spider 实例化 pipeline，但连接对象不是线程安全的，且未管理生命周期容易泄漏。正确姿势是在 open_spider 中初始化 pymongo.MongoClient 并设置 maxPoolSize，在 close_spider 中显式调用 client.close()。
-
maxPoolSize建议设为 5–10，和并发请求数匹配，避免连接耗尽 - 不要用
connect=False参数跳过初始连接检测，否则第一个insert可能阻塞并超时 - 务必在
close_spider里 flush 剩余 buffer 并 close client，否则进程退出时连接不释放
批量写入必须处理重复、缺失字段和 WriteConcern
insert_many 遇到重复 _id 默认抛 bulk_write_exception，而 Scrapy pipeline 里没 try/catch 就会导致整个 spider crash。同时，不同 spider yield 的 item 字段常不一致，直接 insert_many 可能因某些字段缺失触发 schema validation 失败（如果启用了 strict validation）。
华为云 Python SDK 查询工具。自动发现已安装 SDK 的服务包、Client类、API 方法、入参详情、类信息（__init__ 参数、属性、类型、描述），支持关键字搜索，快速定位目标 API 使用方法。
- 加
ordered=False让失败项不影响其余插入 - 用
upsert=True+ReplaceOne替代纯 insert，避免重复主键错误 - 写入前统一补默认字段（如
updated_at），用item.get('field', default)而非直接取值 - 生产环境建议设
write_concern={'w': 1}，w:2或更高会明显拉长延迟
使用 reactor.callLater 替代 time.sleep 做定时 flush
在 Twisted 环境下用 time.sleep 会彻底阻塞 reactor，所有请求暂停。必须用 Scrapy 原生支持的 reactor.callLater 注册延迟回调，它把 flush 任务扔进事件循环队列，不影响调度。
示例关键逻辑：
def open_spider(self, spider):
self.buffer = []
self.timer = None
# …… client 初始化
self.flush_interval = 5 # 秒
<p>def process_item(self, item, spider):
self.buffer.append(dict(item))
if len(self.buffer) >= self.batch_size:
self._flush(spider)
elif self.timer is None:
self.timer = reactor.callLater(self.flush_interval, self._flush, spider)
return item</p><p>def _flush(self, spider):
if not self.buffer:
return
try:
self.collection.insert_many(self.buffer, ordered=False)
except Exception as e:
spider.logger.error(f"Mongo flush failed: {e}")
finally:
self.buffer.clear()
self.timer = None</p>
注意 callLater 返回的 timer 对象要保存，避免被 GC；每次 flush 后清空 timer，防止重复触发。buffer 清空前若发生异常，当前批次数据就丢了——这是异步批量写入无法完全规避的风险点，业务上需权衡是否加本地日志兜底。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
