快捷導(dǎo)航

Python多進(jìn)程協(xié)作模擬實(shí)現(xiàn)流程

更新時(shí)間：2023年01月17日 10:15:46 作者：永遠(yuǎn)的麥田

當(dāng)多個(gè)進(jìn)程使用同一份數(shù)據(jù)資源的時(shí)候，因?yàn)檫M(jìn)程的運(yùn)行沒有順序，運(yùn)行起來也無法控制，如果不加以干預(yù)，往往會(huì)引發(fā)數(shù)據(jù)安全或順序混亂的問題，所以要在多個(gè)進(jìn)程讀寫共享數(shù)據(jù)資源的時(shí)候加以適當(dāng)?shù)牟呗?，來保證數(shù)據(jù)的一致性問題

由于python多線程無法發(fā)揮多核的作用，因此當(dāng)計(jì)算量很大的時(shí)候就需要考慮多進(jìn)程。

只不過多進(jìn)程比較麻煩一些，進(jìn)程中通信向來是一件麻煩事。

python提供了multiprocessing 模塊，應(yīng)該會(huì)提供不少便利

假設(shè)我們做量化交易需要運(yùn)行一個(gè)算法，針對(duì)一只票每運(yùn)行一次需要30秒，運(yùn)行完成將結(jié)果作為一個(gè)因子交給另一個(gè)計(jì)算模塊運(yùn)行，最后將計(jì)算結(jié)果統(tǒng)計(jì)出來。如果只運(yùn)行一只票問題不大，大不了等一下就好，同時(shí)運(yùn)行10票的話就很有必然考慮多進(jìn)程以加快速度。

pro_queue為生產(chǎn)進(jìn)程的數(shù)據(jù)隊(duì)列，生產(chǎn)進(jìn)程完成生產(chǎn)后將自己相關(guān)數(shù)據(jù)裝產(chǎn)pro_queue中

calc_queue為計(jì)算進(jìn)程的數(shù)據(jù)隊(duì)列，計(jì)算進(jìn)程先將生產(chǎn)進(jìn)程的數(shù)據(jù)通過參數(shù)傳遞的方式獲取到，然后進(jìn)行計(jì)算，并將計(jì)算結(jié)果存入calc_queue中

主進(jìn)程在最后不斷獲取計(jì)算進(jìn)程的計(jì)算結(jié)果，并將其累加，最終獲取到一個(gè)總的計(jì)算結(jié)果

生產(chǎn)進(jìn)程：

def func_a(i, p: queues):
    print(F"第{i}個(gè)生產(chǎn)進(jìn)程: 進(jìn)程號(hào):{os.getpid()},開始...")
    time.sleep(random.randint(1, 5))
    result = random.randint(50, 100)
    print(F"第{i}個(gè)生產(chǎn)進(jìn)程: 完成。生產(chǎn)結(jié)果[{result}]。")
    p.put((i, os.getpid(), result))

計(jì)算進(jìn)程：

def func_b(i, data, q: queues):
    print(F"第{i}個(gè)計(jì)算進(jìn)程: 進(jìn)程號(hào):{os.getpid()}, 數(shù)據(jù)源{data[0]}:{data[1]}:{data[2]},開始...")
    time.sleep(random.randint(1, 3))
    result = data[2] * 100
    print(F"第{i}個(gè)計(jì)算進(jìn)程: 完成。計(jì)算結(jié)果【{result}】")
    q.put(result)

程序一開始先將生產(chǎn)進(jìn)程啟動(dòng)起來：其中process_num表示同時(shí)啟動(dòng)多少個(gè)生產(chǎn)進(jìn)程

    for i in range(process_num):
        p = Process(target=func_a, args=(i, pro_queue,))
        p.start()

然后不停等待生產(chǎn)進(jìn)程的結(jié)果，每等到一個(gè)結(jié)果就啟動(dòng)一個(gè)計(jì)算進(jìn)程

    for i in range(process_num):
        data = pro_queue.get()
        p = Process(target=func_b, args=(i, data, calc_queue))
        p.start()

最后就是等待計(jì)算進(jìn)程的計(jì)算結(jié)果

    s = 0
    for i in range(process_num):
        s += calc_queue.get()
    print(F"所有計(jì)算完成，總結(jié)果：{s}")

補(bǔ)上程序一開始的兩個(gè)隊(duì)列的申明：

import multiprocessing
import time
from multiprocessing import Process
from multiprocessing import queues
import random
import os
    pro_queue = queues.Queue(ctx=multiprocessing)
    calc_queue = queues.Queue(ctx=multiprocessing)
    process_num = 5

到此這篇關(guān)于Python多進(jìn)程協(xié)作模擬實(shí)現(xiàn)流程的文章就介紹到這了,更多相關(guān)Python多進(jìn)程協(xié)作內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: