本文和大家分享的主要是python 連接數(shù)據(jù)庫(kù)異步存儲(chǔ)相關(guān)內(nèi)容，一起來(lái)看看吧，希望對(duì)大家 學(xué)習(xí)python有所幫助。

　　當(dāng)同步寫(xiě)入數(shù)據(jù)庫(kù)時(shí), 可能會(huì)發(fā)生下載速度很快 , 但是寫(xiě)入速度很慢的情況 , 我們采用異步存儲(chǔ)寫(xiě)入數(shù)據(jù)庫(kù) .

　　 實(shí)現(xiàn)異步寫(xiě)入mysql數(shù)據(jù)庫(kù)的思路：

　　1, 將數(shù)據(jù)庫(kù)的連接數(shù)據(jù)寫(xiě)入到 settings 文件中，供后面自定義的 pipeline 使用

　　2, 自定義 pipeline, 使用 Twisted 框架實(shí)現(xiàn)異步

　　3, 在 settings 中注冊(cè)這個(gè)管道

　　在settings 設(shè)置連接信息

　　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 自定義一個(gè)新的類(lèi)來(lái)實(shí)現(xiàn)異步存儲(chǔ)

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

　　 def  __init__(self,dbpool):

　　# 定義連接池為對(duì)象函數(shù)

　　self.dbpool=dbpool

　　query = self.dbpool.runInteraction(self.create_table)

　　# 創(chuàng)建存儲(chǔ)表

　　 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)

　　# 自動(dòng)調(diào)用的 , 只調(diào)用一次

　　# 從 settings.py 中根據(jù)字段加載對(duì)應(yīng)的文件     @classmethod

　　 def  from_settings(cls, settings):

　　#POOL 池子 ,con: 連接

　　# 參數(shù) 1:dbapiName  數(shù)據(jù)庫(kù)接口名稱

　　# 參數(shù) 2:*connargs *args

　　# 參數(shù) 3:*connkw **kwargs

　　# 將 setting 中連接數(shù)據(jù)庫(kù)所需內(nèi)容取出

　　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,

　　)

　　# 通過(guò) Twisted 框架提供的容器連接數(shù)據(jù)庫(kù)

　　dbpool = adbapi.ConnectionPool(

　　"MySQLdb", **config

　　)

　　#cls 把參數(shù)給 __init__

　　 return cls(dbpool)

　　#roll back: 回滾

　　#commit: 提交

　　# 事務(wù) : 如果所有語(yǔ)句都執(zhí)行正確 , 才真正執(zhí)行 , 只要有一條數(shù)據(jù)出錯(cuò) , 可以通過(guò)回滾撤銷(xiāo)所有操作

　　# 開(kāi)啟事務(wù)

　　#  嘗試執(zhí)行多條 sql 語(yǔ)句

　　# 沒(méi)問(wèn)題  commit

　　# 有問(wèn)題  roll back

　　# 關(guān)閉事務(wù)

　　 def  process_item(self, item, spider):

　　#runInteraction: 運(yùn)行交互

　　query =self.dbpool.runInteraction(self.insert_sql,item)

　　# 當(dāng)執(zhí)行過(guò)程中出現(xiàn)錯(cuò)誤 , 執(zhí)行 adderrback

　　query.addErrback(self.insert_error, item, spider)

　　 return item

　　 def  insert_error(self,failed):

　　 print ">>>>>>>>>>>",failed

　　 def  insert_sql(self,cursor, item):

　　# 執(zhí)行具體的插入語(yǔ)句 , 不需要 commit 操作 ,Twisted 會(huì)自動(dòng)進(jìn)行

　　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 里注冊(cè) :

　　ITEM_PIPELINES = {

　　#Mysql 異步寫(xiě)入

　　"JobboleSpider.pipelines.MySQLAsynPipeline": 2,

　　}

　　完成異步存儲(chǔ), 數(shù)據(jù)存往數(shù)據(jù)庫(kù)的速度與下載的速度差不多快

來(lái)源： 博客園