思路：

隊(duì)列使用說(shuō)明：

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

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

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

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

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

5.依次啟動(dòng)

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

代碼如下

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ù)庫(kù)連接 # 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)錯(cuò)誤,非待支付訂單’ 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ù)量來(lái)創(chuàng)建線程t = Thread(target=self.asynchronousPay,args=(k[0],k[1]))threads.append(t) for s in nloops: # 開(kāi)始多線程threads[s].start() for j in nloops: # 等待所有線程完成threads[j].join() else: print('隊(duì)列數(shù)量為空')if __name__=='__main__': start_time = time.time() # 計(jì)算程序開(kāi)始時(shí)間 wechfy=doWeChatNotify() wechfy.run_multithreading()#多線程 print(’程序耗時(shí){:.2f}’.format(time.time() - start_time)) # 計(jì)算程序總耗時(shí)

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

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