Python Scrapy结合adbapi实现异步高效数据库写入

Noah

发布时间：2026-10-14 | 4人浏览过

原创

本文介绍如何在Scrapy框架中利用Twisted库的adbapi模块实现数据库的异步写入。通过建立连接池和runInteraction方法，解决同步提交导致的阻塞问题，显著提升数据解析与写入的效率，适用于高并发爬虫场景。

一：twisted中的adbapi

在使用pymysql进行数据库操作时，commit()和execute()方法默认执行同步提交。由于Scrapy框架基于异步多线程进行数据解析，其解析速度往往远快于数据库写入速度。若写入过程过慢，极易造成数据库写入阻塞，进而严重影响整体写入效率。为解决这一问题，可以利用Twisted异步IO框架实现数据的异步写入。通过多线程异步机制处理数据写入，能够有效提升数据库的写入速度。

1.1 两个主要方法

adbapi.ConnectionPool：

该方法用于创建一个数据库连接池对象，池中包含多个连接对象，每个连接均在独立的线程中运行。需要注意的是，adbapi仅提供异步访问数据库的编程框架，其底层依然依赖如MySQLdb等库来实际访问数据库。

dbpool.runInteraction(do_insert,item)：

此方法用于异步调用do_insert函数。dbpool会从连接池中选取一个连接对象，在独立线程中执行insert_db操作。其中，参数item将传递给do_insert的第二个参数，而第一个参数则是一个Transaction对象。该对象接口与Cursor类似，支持调用execute方法执行SQL语句。do_insert执行完毕后，连接对象会自动调用commit方法进行提交。

1.2 使用实例


from twisted.enterprise import adbapi


# 初始化数据库连接池(线程池)
# 参数一：mysql的驱动
# 参数二：连接mysql的配置信息
dbpool = adbapi.ConnectionPool('pymysql', **params)


# 参数1：在异步任务中要执行的函数insert_db；
# 参数2：给该函数insert_db传递的参数
query = self.dbpool.runInteraction(self.do_insert, item)


# 在execute()之后，不需要再进行commit()，连接池内部会进行提交的操作。
def do_insert(self, cursor, item):
    insert_sql = """
            insert into qa_sample( 
            need_id, 
            need_question_uptime, 
            need_title, 
            need_title_describe, 
            need_answer_uptime, 
            need_answer)
            values (%s, %s, %s, %s, %s, %s)
            """
    params = (item['need_id'],
              item['need_question_uptime'],
              item['need_title'],
              item['need_title_describe'],
              item['need_answer_uptime'],
              item['need_answer'])
    cursor.execute(insert_sql, params)

二：结合scrapy中的pipelines


# -*- coding: utf-8 -*-
from twisted.enterprise import adbapi
import pymysql
 
# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: https://docs.scrapy.org/en/latest/topics/item-pipeline.html
 
 
class QaSpiderPipeline(object):
    def process_item(self, item, spider):
        return item
 
class MysqlTwistedPipeline(object):
    def __init__(self, dbpool):
        self.dbpool = dbpool
 
    @classmethod
    def from_settings(cls, settings):
        dbparams = dict(
            host=settings['MYSQL_HOST'],
            db=settings['MYSQL_DBNAME'],
            user=settings['MYSQL_USER'],
            passwd=settings['MYSQL_PASSWORD'],
            charset='utf8',
            cursorclass=pymysql.cursors.DictCursor,
            use_unicode=True
        )
        dbpool = adbapi.ConnectionPool('pymysql', **dbparams)
        return cls(dbpool)
 
    def process_item(self, item, spider):
        query = self.dbpool.runInteraction(self.do_insert, item)
 
    def do_insert(self, cursor, item):
        insert_sql = """
                insert into qa_sample( 
                need_id, 
                need_question_uptime, 
                need_title, 
                need_title_describe, 
                need_answer_uptime, 
                need_answer)
                values (%s, %s, %s, %s, %s, %s)
                """
        params = (item['need_id'],
                  item['need_question_uptime'],
                  item['need_title'],
                  item['need_title_describe'],
                  item['need_answer_uptime'],
                  item['need_answer'])
        cursor.execute(insert_sql, params)

到此这篇关于Python中Scrapy+adbapi提高数据库写入效率实现的文章就介绍到这了,更多相关Scrapy+adbapi数据库写入内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站！