python程序中的線程操作 concurrent模塊使用詳解

更新時(shí)間：2019年09月23日 08:40:31 作者：Cecilia喜陳

這篇文章主要介紹了python程序中的線程操作 concurrent模塊使用詳解,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

一、concurrent模塊的介紹

concurrent.futures模塊提供了高度封裝的異步調(diào)用接口

ThreadPoolExecutor：線程池，提供異步調(diào)用

ProcessPoolExecutor：進(jìn)程池，提供異步調(diào)用

ProcessPoolExecutor 和 ThreadPoolExecutor：兩者都實(shí)現(xiàn)相同的接口，該接口由抽象Executor類定義。

二、基本方法

submit(fn, *args, **kwargs) :異步提交任務(wù)

map(func, *iterables, timeout=None, chunksize=1) ：取代for循環(huán)submit的操作

shutdown(wait=True) ：相當(dāng)于進(jìn)程池的pool.close()+pool.join()操作

wait=True，等待池內(nèi)所有任務(wù)執(zhí)行完畢回收完資源后才繼續(xù)
wait=False，立即返回，并不會等待池內(nèi)的任務(wù)執(zhí)行完畢
但不管wait參數(shù)為何值，整個(gè)程序都會等到所有任務(wù)執(zhí)行完畢
submit和map必須在shutdown之前

result(timeout=None) ：取得結(jié)果

add_done_callback(fn) ：回調(diào)函數(shù)

三、進(jìn)程池和線程池

池的功能：限制進(jìn)程數(shù)或線程數(shù).

什么時(shí)候限制：當(dāng)并發(fā)的任務(wù)數(shù)量遠(yuǎn)遠(yuǎn)大于計(jì)算機(jī)所能承受的范圍,即無法一次性開啟過多的任務(wù)數(shù)量我就應(yīng)該考慮去限制我進(jìn)程數(shù)或線程數(shù),從保證服務(wù)器不崩.

3.1 進(jìn)程池

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Process,current_process
import time
def task(i):
  print(f'{current_process().name} 在執(zhí)行任務(wù){(diào)i}')
  time.sleep(1)
if __name__ == '__main__':
  pool = ProcessPoolExecutor(4) # 進(jìn)程池里又4個(gè)進(jìn)程
  for i in range(20): # 20個(gè)任務(wù)
    pool.submit(task,i)# 進(jìn)程池里當(dāng)前執(zhí)行的任務(wù)i，池子里的4個(gè)進(jìn)程一次一次執(zhí)行任務(wù)

3.2 線程池

from concurrent.futures import ThreadPoolExecutor
from threading import Thread,currentThread
import time
def task(i):
  print(f'{currentThread().name} 在執(zhí)行任務(wù){(diào)i}')
  time.sleep(1)
if __name__ == '__main__':
  pool = ThreadPoolExecutor(4) # 進(jìn)程池里又4個(gè)線程
  for i in range(20): # 20個(gè)任務(wù)
    pool.submit(task,i)# 線程池里當(dāng)前執(zhí)行的任務(wù)i，池子里的4個(gè)線程一次一次執(zhí)行任務(wù)

四、Map的用法

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os,time,random
def task(n):
  print('%s is runing' %os.getpid())
  time.sleep(random.randint(1,3))
  return n**2
if __name__ == '__main__':
  executor=ThreadPoolExecutor(max_workers=3)
  # for i in range(20):
  #   future=executor.submit(task,i)
  executor.map(task,range(1,21)) #map取代了for+submit

五、同步和異步

理解為提交任務(wù)的兩種方式

同步: 提交了一個(gè)任務(wù),必須等任務(wù)執(zhí)行完了(拿到返回值),才能執(zhí)行下一行代碼

異步: 提交了一個(gè)任務(wù),不要等執(zhí)行完了,可以直接執(zhí)行下一行代碼.

同步：相當(dāng)于執(zhí)行任務(wù)的串行執(zhí)行

異步

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Process,current_process
import time
n = 1
def task(i):
  global n
  print(f'{current_process().name} 在執(zhí)行任務(wù){(diào)i}')
  time.sleep(1)
  n += i
  return n
if __name__ == '__main__':
  pool = ProcessPoolExecutor(4) # 進(jìn)程池里又4個(gè)線程
  pool_lis = []
  for i in range(20): # 20個(gè)任務(wù)
    future = pool.submit(task,i)# 進(jìn)程池里當(dāng)前執(zhí)行的任務(wù)i，池子里的4個(gè)線程一次一次執(zhí)行任務(wù)
    # print(future.result()) # 這是在等待我執(zhí)行任務(wù)得到的結(jié)果，如果一直沒有結(jié)果，這里會導(dǎo)致我們所有任務(wù)編程了串行
                # 在這里就引出了下面的pool.shutdown()方法
    pool_lis.append(future)
  pool.shutdown(wait=True) # 關(guān)閉了池的入口，不允許在往里面添加任務(wù)了，會等帶所有的任務(wù)執(zhí)行完，結(jié)束阻塞
  for p in pool_lis:
    print(p.result())
  print(n)# 這里一開始肯定是拿到0的，因?yàn)槲抑皇侨ジ嬖V操作系統(tǒng)執(zhí)行子進(jìn)程的任務(wù)，代碼依然會繼續(xù)往下執(zhí)行
  # 可以用join去解決，等待每一個(gè)進(jìn)程結(jié)束后，拿到他的結(jié)果

六、回調(diào)函數(shù)

import time
from threading import Thread,currentThread
from concurrent.futures import ThreadPoolExecutor
def task(i):
  print(f'{currentThread().name} 在執(zhí)行{i}')
  time.sleep(1)
  return i**2

# parse 就是一個(gè)回調(diào)函數(shù)
def parse(future):
  # 處理拿到的結(jié)果
  print(f'{currentThread().name} 結(jié)束了當(dāng)前任務(wù)')
  print(future.result())
if __name__ == '__main__':
  pool = ThreadPoolExecutor(4)
  for i in range(20):
    future = pool.submit(task,i)
    '''
    給當(dāng)前執(zhí)行的任務(wù)綁定了一個(gè)函數(shù)，在當(dāng)前任務(wù)結(jié)束的時(shí)候就會觸發(fā)這個(gè)函數(shù)（稱之為回調(diào)函數(shù)）
    會把future對象作為參數(shù)傳給函數(shù)
    注：這個(gè)稱為回調(diào)函數(shù)，當(dāng)前任務(wù)處理結(jié)束了，就回來調(diào)parse這個(gè)函數(shù)
    '''
    future.add_done_callback(parse)
    # add_done_callback (parse) parse是一個(gè)回調(diào)函數(shù)
    # add_done_callback () 是對象的一個(gè)綁定方法，他的參數(shù)就是一個(gè)函數(shù)

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

您可能感興趣的文章: