您好,登錄后才能下訂單哦!
這篇文章主要講解了Python多線程實(shí)現(xiàn)支付模擬請求的方法,內(nèi)容清晰明了,對此有興趣的小伙伴可以學(xué)習(xí)一下,相信大家閱讀完之后會有幫助。
思路:
隊(duì)列使用說明:
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)計(jì)函數(shù)運(yùn)行時(shí)長
代碼如下
import asyncio import sys from queue import Queue sys.path.append("../") from tool.__init__ import * from tool.decorator_token import * import time from threading import Thread,Lock class 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_id\t"+"order_sn\t\n" #文件首行數(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_sn with open(DoWeChatNotify_file,'a',encoding='utf-8') as file2:#文件寫入 str_DoWeChatNotifyInfo=str(fwh_order_id)+'\t'+str(fwh_order_sn)+'\t\n' 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:4 timestamp:157129653969 sign: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ù)量來創(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() # 計(jì)算程序開始時(shí)間 wechfy=doWeChatNotify() wechfy.run_multithreading()#多線程 print('程序耗時(shí){:.2f}'.format(time.time() - start_time)) # 計(jì)算程序總耗時(shí)
看完上述內(nèi)容,是不是對Python多線程實(shí)現(xiàn)支付模擬請求的方法有進(jìn)一步的了解,如果還想學(xué)習(xí)更多內(nèi)容,歡迎關(guān)注億速云行業(yè)資訊頻道。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。