Python中Scrapy+adbapi提高数据库写入效率实现

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)

# -*- 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)