Python爬虫：Scrapy使用adbapi提高存储数据库效率

运行环境 Runtime environment

操作系统： Windos10  
IDE: pycharm 2021.3.1 x64  
语言: python v3.9.1
框架: Scrapy

背景

Scrapy 是一个高性能的异步采集框架，采集速度非常快，但是数据库入库受限于数据库本身的原因，一直是个瓶颈。

在某些爬虫项目下，我们需要每次执行一条插入语句，就立即调用commit方法更新数据库。

如果爬取时间太长，中途可能被迫中断，这样程序就不能执行close_spider中的commit。

但如果在insert_db中直接加入commit，又会使程序执行变得很慢。

这里就可以使用Twisted中提供的以异步方式多线程访问数据库的模块adbapi，可以显著提供程序访问数据库的效率。

要点

adbapi.ConnectionPool方法

可以创建一个数据库连接池对象，其中包括多个连接对象，每个连接对象在独立的线程中工作。

adbapi只是提供了异步访问数据库的编程框架，再其内部依然使MySQLdb这样的库访问数据库。
dbpool.runInteraction(insert_db,item)

以异步方式调用insert_db函数，dbpool会选择连接池中的一个连接对象在独立线程中调用insert_db，

其中参数item会被传给insert_db的第二个参数，传给insert_db的第一个参数是一个Transaction对象，

其接口与Cursoru游标对象类似，

可以调用execute方法执行SQL语句，insert_db执行后，连接对象会自动调用commit方法

代码

import scrapy
import pymysql
from itemadapter import ItemAdapter
from twisted.enterprise import adbapi


class MysqlSpiderBasePipeline:
    def __init__(self, dbpool, item_tabla_map):
        self.dbpool = dbpool

    @classmethod
    def from_crawler(cls, cralwer):
        db_parmars = {
            'host': cralwer.settings['MYSQL_HOST'],
            'user': cralwer.settings['MYSQL_USER'],
            'passwd': cralwer.settings['MYSQL_PWD'],
            'db': cralwer.settings['MYSQL_DB'],
            'port': cralwer.settings['MYSQL_PORT'],
            'charset': cralwer.settings['MYSQL_CHARSET']
        }
        dbpool = adbapi.ConnectionPool('pymysql', **db_parmars)
        return cls(dbpool)

    def process_item(self, item, spider):
        # 入库
        query = self.dbpool.runInteraction(
            self.insert_data_to_mysql,
            item
        )
        query.addErrback(
            self.insert_err,
            item
        )
        return item

    def insert_err(self, failure, item):
        print(failure, '失败')  # , item)

    def insert_data_to_mysql(self, cursor, item):
        pass



# 继承   MysqlSpiderBasePipeline 类 然后重写insert_data_to_mysql方法来实现具体的入库逻辑即可。
class SpiderPipeline(MysqlSpiderBasePipeline):
    def insert_data_to_mysql(self, cursor, item):
        sql, data = item.get_update_sql()
        cursor.execute(sql, data)
        # ....