思路：

隊(duì)列使用說明：

　multiprocessing.Queue()#用于進(jìn)程間通信，單主進(jìn)程與子進(jìn)程無法通信（使用進(jìn)程池時盡量不要使用這個） 　multiprocessing.Manager().Queue()#用于主子進(jìn)程通信，通過進(jìn)程池（pool）創(chuàng)建的進(jìn)程可以數(shù)據(jù)共享 　queue.Queue()#用于線程間通信，同一進(jìn)程內(nèi)的數(shù)據(jù)可以共享

1.從數(shù)據(jù)庫里獲取待支付的訂單

2.將獲取出來的數(shù)據(jù)添加至隊(duì)列（queue.Queue()），并在函數(shù)中返回消息隊(duì)列的長度

3.根據(jù)隊(duì)列長度創(chuàng)建對應(yīng)的線程數(shù)量

4.把創(chuàng)建的線程放在list

5.依次啟動

6.最后等待主線程執(zhí)行完結(jié)束，統(tǒng)計函數(shù)運(yùn)行時長

代碼如下

import asyncioimport sysfrom queue import Queuesys.path.append('../')from tool.__init__ import *from tool.decorator_token import *import timefrom threading import Thread,Lockclass doWeChatNotify(BaseTest): def __init__(self): super().__init__() self.limit_num=100 #查詢記錄條數(shù) self.WeChatNotify_sql=’’’select order_id,order_sn from fw_order where `status`=0 and course_id=1569 ORDER BY create_time desc limit %d ;’’’%(self.limit_num) self.fwh_test_api=fwh_test_api self.data = self.my_op.sql_operation_fwh(self.WeChatNotify_sql) self.fwh_order_dict = {} self.que = Queue() @token_fwh#驗(yàn)證token有效性 def get_fwh_token_list(self): token_list=self.fwh_token.loadTokenList() return token_list @token_crm#驗(yàn)證token有 def get_crm_token_list(self) token_list=self.token.loadTokenList() return token_list def testDoWeChatNotify(self): DoWeChatNotify_file=’../tokenFileAndtxtFiles’+’/’+'DoWeChatNotify_asynchronousPay.txt' with open(DoWeChatNotify_file,’a’,encoding=’utf=-8’) as file: str_first='order_idt'+'order_sntn' #文件首行數(shù)據(jù) file.write(str_first) fwh_order_id_list, fwh_order_sn_list = [], [] if self.data!=(): for a in self.data:fwh_order_id=a[’order_id’]fwh_order_sn=a[’order_sn’]self.fwh_order_dict[fwh_order_id]=fwh_order_snwith open(DoWeChatNotify_file,’a’,encoding=’utf-8’) as file2:#文件寫入 str_DoWeChatNotifyInfo=str(fwh_order_id)+’t’+str(fwh_order_sn)+’tn’ file2.flush() #清除緩沖區(qū) file2.write(str_DoWeChatNotifyInfo)self.que.put(self.fwh_order_dict)#將數(shù)據(jù)添加至隊(duì)列 #關(guān)閉數(shù)據(jù)庫連接 # self.my_op.close_db_fwh() # self.my_op.close_db() return self.que.qsize()#返回隊(duì)列數(shù)量 def asynchronousPay(self,order_id,order_sn): count=1 count_num=50 token_list=self.get_fwh_token_list() if (self.data!=()): headers_form_urlencoded[’token’]=token_list[0] url_wechat_success_huidiao=self.fwh_test_api+’/index/Order/doWeChatNotify’ data_wechat_success_huidiao=self.data_to_str.requestDataToStr_firefoxAndChrome_fwh(’’’order_sn:{}order_id:{}meth_id:4timestamp:157129653969sign:0687b01b300b9e300d3996a9d2173f1380973e5a’’’.format(order_sn,order_id)) request_wechat_success_huidiao=requests.post(url=url_wechat_success_huidiao,headers=headers_form_urlencoded,data=data_wechat_success_huidiao) response_wechat_success_huidiao=request_wechat_success_huidiao.json() if ’訂單狀態(tài)錯誤,非待支付訂單’ in response_wechat_success_huidiao[’msg’]:print(data_wechat_success_huidiao) else: print(’待支付訂單為空’) def run_multithreading(self):#多線程 threads = []#存放所有的線程 nloops = list(range(self.testDoWeChatNotify()))#獲取隊(duì)列數(shù)量 if len(nloops)>0: for i,k in zip(nloops,self.que.get().items()):#根據(jù)隊(duì)列數(shù)量來創(chuàng)建線程t = Thread(target=self.asynchronousPay,args=(k[0],k[1]))threads.append(t) for s in nloops: # 開始多線程threads[s].start() for j in nloops: # 等待所有線程完成threads[j].join() else: print('隊(duì)列數(shù)量為空')if __name__=='__main__': start_time = time.time() # 計算程序開始時間 wechfy=doWeChatNotify() wechfy.run_multithreading()#多線程 print(’程序耗時{:.2f}’.format(time.time() - start_time)) # 計算程序總耗時

總結(jié)：親測運(yùn)行時間還是會快很多，單線程支付100個訂單四十幾秒的樣子，多線程運(yùn)行不用join2.x秒，用join八秒的樣子，還有很大的優(yōu)化空間，因?yàn)檫\(yùn)行時會創(chuàng)建100個線程

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。