線程

• 隊列：先進(jìn)先出
• 堆棧：后進(jìn)先出
• 優(yōu)先級：數(shù)字越小優(yōu)先級越大，越先輸出

import queueq = queue.Queue(3) # 先進(jìn)先出-->隊列 q.put('first') q.put(2) # q.put('third') # q.put(4) #由于沒有人取走，就會卡主 q.put(4,block=False) #等同于q.get_nowait(), Ture 阻塞，Flase不阻塞，報異常滿了 # # q.put(4,block=True,timeout=3)print(q.get()) print(q.get()) print(q.get()) print(q.get(block=True,timeout=3)) # 阻塞等待3秒 沒有取走數(shù)據(jù)就報異常 # print(q.get(block=False)) #等同于q.get_nowait() # print(q.get_nowait()) q = queue.LifoQueue(3) #后進(jìn)先出-->堆棧 q.put('first') q.put(2) q.put('third')print(q.get()) print(q.get()) print(q.get()) ''' 打印結(jié)果: third 2 first '''q = queue.PriorityQueue(3) #優(yōu)先級隊列 q.put((10,'one')) q.put((40,'two')) q.put((30,'three'))print(q.get()) print(q.get()) print(q.get()) ''' 數(shù)字越小優(yōu)先級越高 打印結(jié)果 (10, 'one') (30, 'three') (40, 'two') ''' 線程queue

進(jìn)程池線程池

• 池：是用來對進(jìn)程(線程)的數(shù)量加以限制
• 進(jìn)程池：計算密集型，用多進(jìn)程
• 線程池：IO密集型，用多線程，例如：sockect網(wǎng)絡(luò)通信就應(yīng)該用多線程

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor import os,time,random''' sockect網(wǎng)絡(luò)通信是IO操作，所以用多線程 計算密集型：用多進(jìn)程 '''def task(name):print('name:%s pid:%s run' %(name,os.getpid()))time.sleep(random.randint(1,3))if __name__ == '__main__':# pool = ProcessPoolExecutor(4) # 進(jìn)程池最多裝4個進(jìn)程,不指定的話默認(rèn)是cpu的核數(shù)pool = ThreadPoolExecutor(5)for i in range(10):pool.submit(task,'yang%s' %i) # 異步調(diào)用池子收了10個任務(wù)，但同一時間只有4個任務(wù)在進(jìn)行 pool.shutdown(wait=True) # 類似join 代表往池子里面丟任務(wù)的入口關(guān)掉 計數(shù)器-1print('主') ''' 打印結(jié)果： name:yang0 pid:11120 run name:yang1 pid:11120 run name:yang2 pid:11120 run name:yang3 pid:11120 run name:yang4 pid:11120 runname:yang5 pid:11120 run name:yang6 pid:11120 run name:yang7 pid:11120 runname:yang8 pid:11120 run name:yang9 pid:11120 run 主 '''from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor from threading import currentThread import os,time,randomdef task():print('name:%s pid:%s run' %(currentThread().getName(),os.getpid()))time.sleep(random.randint(1,3))if __name__ == '__main__':# pool = ProcessPoolExecutor(4) # 進(jìn)程池最多裝4個進(jìn)程,不指定的話默認(rèn)是cpu的核數(shù)pool = ThreadPoolExecutor(5)for i in range(10):pool.submit(task) # 異步調(diào)用池子收了10個任務(wù)，但同一時間只有4個任務(wù)在進(jìn)行 pool.shutdown(wait=True) # 類似join 代表往池子里面丟任務(wù)的入口關(guān)掉 計數(shù)器-1print('主') ''' 打印結(jié)果： name:ThreadPoolExecutor-0_0 pid:14052 run name:ThreadPoolExecutor-0_1 pid:14052 run name:ThreadPoolExecutor-0_2 pid:14052 run name:ThreadPoolExecutor-0_3 pid:14052 run name:ThreadPoolExecutor-0_4 pid:14052 run name:ThreadPoolExecutor-0_2 pid:14052 run name:ThreadPoolExecutor-0_1 pid:14052 run name:ThreadPoolExecutor-0_3 pid:14052 run name:ThreadPoolExecutor-0_4 pid:14052 run name:ThreadPoolExecutor-0_0 pid:14052 run 主 ''' 進(jìn)程池|線程池

同步調(diào)用和異步調(diào)用

提交任務(wù)的兩種方式：

• 同步調(diào)用:提交完任務(wù)后，就在原地等待任務(wù)執(zhí)行完畢，拿到結(jié)果，再執(zhí)行下一行代碼,導(dǎo)致程序是串行執(zhí)行
• 異步調(diào)用:提交完任務(wù)后，不在原地等待任務(wù)執(zhí)行完。回調(diào)機制：自動觸發(fā)

#1.同步調(diào)用:提交完任務(wù)后，就在原地等待任務(wù)執(zhí)行完畢，拿到結(jié)果，再執(zhí)行下一行代碼,導(dǎo)致程序是串行執(zhí)行from concurrent.futures import ThreadPoolExecutor import time import randomdef la(name):print('%s is laing' %name)time.sleep(random.randint(3,5))res = random.randint(7,13)*'#'return {'name':name,'res':res}def weigh(shit):name = shit['name']size = len(shit['res'])print('%s 拉了 <%s>kg' %(name,size))if __name__ == '__main__':pool = ThreadPoolExecutor(10)shit1 = pool.submit(la,'alex').result()weigh(shit1)shit2 = pool.submit(la,'yang').result()weigh(shit2)shit3 = pool.submit(la,'hang').result()weigh(shit3) ''' 打印結(jié)果： alex is laing alex 拉了 <8>kg yang is laing yang 拉了 <8>kg hang is laing hang 拉了 <7>kg ''' 同步調(diào)用 #2.異步調(diào)用:提交完任務(wù)后，不在原地等待任務(wù)執(zhí)行完 from concurrent.futures import ThreadPoolExecutor import time import randomdef la(name):print('%s is laing' %name)time.sleep(random.randint(3,5))res = random.randint(7,13)*'#'return {'name':name,'res':res}# weigh({'name':name,'res':res}) # 這樣寫,所有功能 不能體現(xiàn)出解耦合def weigh(shit):shit = shit.result() # 拿到是一個對象，需要進(jìn)行result()name = shit['name']size = len(shit['res'])print('%s 拉了 <%s>kg' %(name,size))if __name__ == '__main__':pool = ThreadPoolExecutor(10)shit1 = pool.submit(la,'alex').add_done_callback(weigh)shit2 = pool.submit(la,'yang').add_done_callback(weigh)shit3 = pool.submit(la,'hang').add_done_callback(weigh) ''' 打印結(jié)果: alex is laing yang is laing hang is laing hang 拉了 <10>kg alex 拉了 <7>kg yang 拉了 <12>kg ''' 異步調(diào)用

異步調(diào)用的應(yīng)用

from concurrent.futures import ThreadPoolExecutor import requests import timedef get(url):print('GET %s'%url)response = requests.get(url)time.sleep(3)return {'url':url,'content':response.text}def parse(res):res = res.result()print('%s parse res is %s' %(res['url'],len(res['content'])))if __name__ == '__main__':urls = ['http://www.cnblogs.com/linhaifeng','https://www.python.org','https://www.openstack.org',]pool = ThreadPoolExecutor(2)for url in urls:pool.submit(get,url).add_done_callback(parse) ''' 打印結(jié)果： GET http://www.cnblogs.com/linhaifeng GET https://www.python.org http://www.cnblogs.com/linhaifeng parse res is 16320 GET https://www.openstack.org https://www.python.org parse res is 49273 https://www.openstack.org parse res is 64040 ''' 應(yīng)用

轉(zhuǎn)載于:https://www.cnblogs.com/Mryang123/p/8921962.html

總結(jié)

以上是生活随笔為你收集整理的并发编程---线程queue---进程池线程池---异部调用(回调机制)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。