scrapy 2.x 的 pipeline 必须同步，异步写入 mysql 会出错；应使用 twisted.enterprise.adbapi 实现线程池异步操作，process_item 返回 runinteraction 调用，_do_insert 中用 tx.execute 执行 sql，需实现 close_spider 和 adderrback 处理异常。

Scrapy 2.x 的 Pipeline 必须是同步的，异步写入 MySQL 会出错

Scrapy 2.x 的 process_item 方法本身不支持 async/await，直接在 Pipeline 中用 await conn.execute(...) 会导致 RuntimeWarning: coroutine '...' was never awaited，甚至任务静默丢弃。这不是配置问题，而是 Scrapy 2.x 的核心设计限制：整个 Pipeline 链路运行在主线程、同步调用。

用 twisted.enterprise.adbapi 实现真正异步 MySQL 写入

Scrapy 原生推荐且唯一可靠的异步数据库方案是 adbapi —— 它把阻塞的 DB-API 调用扔进线程池执行，再用 Deferred 回调通知结果。你不需要自己管理线程或事件循环。

MySQL

编写正确的MySQL查询，避免字符集、索引和锁方面的常见陷阱。

下载

• 安装依赖：pip install PyMySQL（或 mysqlclient）
• 在 pipelines.py 中初始化连接池：

  from twisted.enterprise import adbapi
  import pymysql
  
  class MysqlPipeline:
      def __init__(self, dbpool):
          self.dbpool = dbpool
  
      @classmethod
      def from_crawler(cls, crawler):
          dbparams = dict(
              host=crawler.settings.get('MYSQL_HOST', 'localhost'),
              port=crawler.settings.get('MYSQL_PORT', 3306),
              user=crawler.settings.get('MYSQL_USER', 'root'),
              password=crawler.settings.get('MYSQL_PASSWORD', ''),
              database=crawler.settings.get('MYSQL_DATABASE', 'scrapy_db'),
              charset='utf8mb4',
              cursorclass=pymysql.cursors.DictCursor,
          )
          dbpool = adbapi.ConnectionPool('pymysql', **dbparams)
          return cls(dbpool)
  

• process_item 返回 self.dbpool.runInteraction(self._do_insert, item)，其中 _do_insert 是普通同步函数，接收 tx（twisted 的 transaction 对象）作为第一个参数

_do_insert 里不能用 await，但可以安全执行原生 SQL

adbapi 的 runInteraction 会在独立线程中调用你的回调函数，所以这里写标准的 tx.execute() 即可，无需考虑协程。注意事务自动管理：成功则提交，异常则回滚。

• 示例插入逻辑：

  def _do_insert(self, tx, item):
      sql = "INSERT INTO books (title, price, url) VALUES (%s, %s, %s)"
      args = (item.get('title'), item.get('price'), item.get('url'))
      tx.execute(sql, args)
  

• 如果需要主键或影响行数，用 tx.rowcount 或 tx.lastrowid（PyMySQL 支持）
• 避免在 _do_insert 中调用其他异步函数、Scrapy 的 Request 或 ItemLoader —— 这些对象不是线程安全的

别漏掉 close_spider 和异常日志，否则连接池不释放、错误被吞

adbapi.ConnectionPool 不会自动关闭，必须显式清理；同时 runInteraction 抛出的异常默认不打印，容易误判数据丢失。

• 在 Pipeline 中添加：

  def close_spider(self, spider):
      self.dbpool.close()
  

• 给 process_item 加 addErrback 捕获数据库异常：

  def process_item(self, item, spider):
      d = self.dbpool.runInteraction(self._do_insert, item)
      d.addErrback(self._handle_error, item, spider)
      return item
  
  def _handle_error(self, failure, item, spider):
      spider.logger.error(f"Database error on {item}: {failure.value}")
  

• 确保 ITEM_PIPELINES 设置启用该 Pipeline，且优先级低于去重、验证类 Pipeline

Scrapy 2.x 的异步数据库写入，本质是「用线程模拟异步」，不是真协程。很多人卡在试图用 asyncio + aiomysql，结果发现 Scrapy 根本不等你 await 完——这点必须提前认清。