scrapy异步存储mysql大量数据库

from twisted.enterprise import adbapi

class MySQLPipeline(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='utf8mb4',
            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)
        query.addErrback(self._handle_error, item, spider)
        return item

    def _do_insert(self, cursor, item):
        # 执行插入操作的SQL语句
        pass

    def _handle_error(self, failure, item, spider):
        # 处理错误
        pass

import pymysql
from twisted.enterprise import adbapi

class MySQLPipeline:
    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='utf8mb4',
            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)
        query.addErrback(self._handle_error, item, spider)
        return item

    def _do_insert(self, cursor, item):
        # 根据item构建SQL语句并执行插入操作
        sql = "INSERT INTO table_name (column1, column2) VALUES (%s, %s)"
        cursor.execute(sql, (item['field1'], item['field2']))

    def _handle_error(self, failure, item, spider):
        # 处理可能出现的错误
        print(failure)