快捷導(dǎo)航

Python使用concurrent.futures模塊實現(xiàn)多進(jìn)程多線程編程

更新時間：2023年12月03日 08:45:01 作者：小龍在山東

Python的concurrent.futures模塊可以很方便的實現(xiàn)多進(jìn)程、多線程運行,減少了多進(jìn)程帶來的的同步和共享數(shù)據(jù)問題,下面就跟隨小編一起了解一下concurrent.futures模塊的具體使用吧

Executor是一個抽象類，表示一個可執(zhí)行的上下文。Future則代表一個將要執(zhí)行的任務(wù)，并提供了一些方法來獲取任務(wù)的狀態(tài)和結(jié)果。ThreadPoolExecutor是Executor的一個具體實現(xiàn)類，它使用線程池來執(zhí)行任務(wù)。

多線程

from concurrent.futures import ThreadPoolExecutor
import time


# 任務(wù)函數(shù)
def task(name):
    print(f"任務(wù){(diào)name}開始執(zhí)行")
    time.sleep(2)
    print(f"任務(wù){(diào)name}執(zhí)行完畢")
    return True


# 創(chuàng)建線程池
executor = ThreadPoolExecutor(max_workers=2)
# 提交任務(wù)
futures = []
for i in range(5):
    future = executor.submit(task, f"任務(wù){(diào)i + 1}")
    futures.append(future)
# 等待所有任務(wù)完成
executor.shutdown()
# 打印任務(wù)結(jié)果
for future in futures:
    print(future.result())

首先創(chuàng)建線程池：ThreadPoolExecutor對象executor，

然后提交任務(wù)列表：submit到線程池返回future，加入任務(wù)列表。

設(shè)置等待所有任務(wù)完成：executor.shutdown()，

最后查看執(zhí)行結(jié)果：future.result()。

多進(jìn)程

這里改用了ProcessPoolExecutor線程池。

import os
import random
import time
from concurrent.futures import ProcessPoolExecutor


def task(n):
    print('%s is runing' % os.getpid())
    time.sleep(random.randint(1, 3))
    return n ** 2


if __name__ == '__main__':
    executor = ProcessPoolExecutor(max_workers=3)

    futures = []
    for i in range(11):
        future = executor.submit(task, i)
        futures.append(future)
    executor.shutdown(True)
    for future in futures:
        print(future.result())

add_done_callback設(shè)置回調(diào)函數(shù)

import os
from concurrent.futures import ProcessPoolExecutor

import requests


def get_page(url):
    print('<進(jìn)程%s> get %s' % (os.getpid(), url))
    respone = requests.get(url)
    if respone.status_code == 200:
        return {'url': url, 'text': respone.text}


def parse_page(res):
    res = res.result()
    print('<進(jìn)程%s> parse %s' % (os.getpid(), res['url']))
    parse_res = 'url:<%s> size:[%s]\n' % (res['url'], len(res['text']))
    with open('db.txt', 'a') as f:
        f.write(parse_res)


if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://help.github.com/',
        'http://www.sina.com.cn/'
    ]

    p = ProcessPoolExecutor(3)
    for url in urls:
        p.submit(get_page, url).add_done_callback(parse_page)
        # parse_page拿到的是一個future對象obj，需要用obj.result()拿到結(jié)果