python连接数据库异步存储

Posted 2020-10-03

当同步写入数据库时,可能会发生下载速度很快,但是写入速度很慢的情况,因此我们采用异步存储写入数据库.     

实现异步写入mysql数据库的思路：    

    1,将数据库的连接数据写入到settings文件中，供后面自定义的pipeline使用
    2,自定义pipeline,使用Twisted框架实现异步
    3,在settings中注册这个管道

在settings设置连接信息

 

1 MYSQL_HOST = ‘127.0.0.1‘
2 MYSQL_DBNAME = ‘testdb‘
3 MYSQL_USER = ‘root‘
4 MYSQL_PASSWD = ‘123456‘
5 MYSQL_CHARSET="UTF8"
6 
7 MYSQL_PORT = 3306

 

在pipeline自定义一个新的类来实现异步存储

from MySQLdb.cursors import DictCursor
from twisted.enterprise import adbapi
class MySQLAsynPipeline(object):

    def __init__(self,dbpool):
        #定义连接池为对象函数
        self.dbpool=dbpool
        query = self.dbpool.runInteraction(self.create_table)
    #创建存储表
    def create_table(self,cursor):
        sql = "create table if not exists test (id INT PRIMARY KEY auto_increment NOT NULL , title VARCHAR(50) NOT NULL,category_name VARCHAR (100),date_time VARCHAR (20) NOT NULL ,likes INT DEFAULT 0,content longtext ,comment INT DEFAULT 0,collect INT DEFAULT 0,detail_url VARCHAR (255) UNIQUE,src VARCHAR (255))"
        cursor.execute(sql)


    #自动调用的,只调用一次
    #从settings.py中根据字段加载对应的文件
    @classmethod
    def from_settings(cls, settings):


        #POOL池子,con:连接
        #参数1:dbapiName 数据库接口名称
        #参数2:*connargs *args
        #参数3:*connkw **kwargs
        #将setting中连接数据库所需内容取出

        config = dict(
            host=settings[‘MYSQL_HOST‘],
            db = settings[‘MYSQL_DBNAME‘],
            user = settings[‘MYSQL_USER‘],
            passwd =settings[‘MYSQL_PASSWD‘],
            charset = settings[‘MYSQL_CHARSET‘],
            port = settings[‘MYSQL_PORT‘],
            cursorclass = DictCursor,
        )
        #通过Twisted框架提供的容器连接数据库
        dbpool = adbapi.ConnectionPool(
            "MySQLdb", **config
        )

        #cls把参数给__init__
        return cls(dbpool)
     #roll back:回滚
     #commit:提交
    #事务:如果所有语句都执行正确,才真正执行,只要有一条数据出错,可以通过回滚撤销所有操作


    #开启事务
    # 尝试执行多条sql语句
    #没问题 commit
    #有问题 roll back
    #关闭事务
    def process_item(self, item, spider):
        #runInteraction:运行交互
        query =self.dbpool.runInteraction(self.insert_sql,item)
        #当执行过程中出现错误,执行adderrback
        query.addErrback(self.insert_error, item, spider)
        return item

    def insert_error(self,failed):
        print ">>>>>>>>>>>",failed

    def insert_sql(self,cursor, item):
        #执行具体的插入语句,不需要commit操作,Twisted会自动进行
        sql = "insert into test (title,category_name, date_time,likes,content, comment,collect, detail_url,src) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)"
        cursor.execute(sql, (
        item[‘title‘], item[‘category_name‘], item[‘date_time‘], item[‘likes‘], item[‘content‘], item[‘comment‘],
        item[‘collect‘], item[‘detail_url‘], item[‘src‘][0])

在settings里注册:

ITEM_PIPELINES = {
    #Mysql异步写入
    "JobboleSpider.pipelines.MySQLAsynPipeline": 2,

}

完成异步存储,数据存往数据库的速度与下载的速度差不多快