Python（29）：Python程序中的進程操作(multiprocess.process)

本文轉載自查看原文 2019-12-09 17:38 498 31 Python

一、multiprocess模塊

multiprocess不是一個模塊而是python中一個操作、管理進程的包。

子模塊分為四個部分：

創建進程部分(multiprocess.process)
進程同步部分((multiprocess.Lock))
進程池部分((multiprocess.Pool）)
進程之間數據共享(ThreadLocal、multiprocess.Queue、Pipes)

二、multiprocess.process模塊

process模塊是一個創建進程的模塊，借助這個模塊，就可以完成進程的創建。

在windows中使用process模塊的注意事項

在Windows操作系統中由於沒有fork(linux操作系統中創建進程的機制)，在創建子進程的時候會自動 import 啟動它的這個文件，而在 import 的時候又執行了整個文件。

因此如果將process()直接寫在文件中就會無限遞歸創建子進程報錯。所以必須把創建子進程的部分使用if \_\_name\_\_ =='\_\_main\_\_' 判斷保護起來，import 的時候，就不會遞歸運行了。

2、使用process模塊創建進程

在一個python進程中開啟子進程，start方法和並發效果。

1 在Python中啟動的第一個子進程

import time
from multiprocessing import Process

def f(name):
    print('hello', name)
    time.sleep(1)
    print('我是子進程')


if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    # p.join()
    print('我是父進程')

2、查看主進程和子進程的進程號

import os
from multiprocessing import Process

def f(x):
    print('子進程id ：',os.getpid(),'父進程id ：',os.getppid())
    return x*x

if __name__ == '__main__':
    print('主進程id ：', os.getpid())
    p_lst = []
    for i in range(5):
        p = Process(target=f, args=(i,))
        p.start()

3、進階，多個進程同時運行

注意，子進程的執行順序不是根據啟動順序決定的。

import time
from multiprocessing import Process


def f(name):
    print('hello', name)
    time.sleep(1)


if __name__ == '__main__':
    p_lst = []
    for i in range(5):
        p = Process(target=f, args=('bob',))
        p.start()
        p_lst.append(p)
        p.join()
     # [p.join() for p in p_lst]
    print('父進程在執行')

4、通過繼承Process類開啟進程

import os
from multiprocessing import Process


class MyProcess(Process):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print(os.getpid())
        print('%s 正在和女主播聊天' %self.name)

p1=MyProcess('wupeiqi')
p2=MyProcess('yuanhao')
p3=MyProcess('nezha')

p1.start() # start會自動調用run
p2.start()
# p2.run()
p3.start()


p1.join()
p2.join()
p3.join()

print('主線程')

5、進程之間的數據隔離問題

from multiprocessing import Process

def work():
    global n n=0
    print('子進程內: ',n)


if __name__ == '__main__':
    n = 100
    p=Process(target=work)
    p.start()
    print('主進程內: ',n)

3、守護進程daemon

會隨着主進程的結束而結束。

主進程創建守護進程

　　其一：守護進程會在主進程代碼執行結束后就終止

　　其二：守護進程內無法再開啟子進程,否則拋出異常：AssertionError: daemonic processes are not allowed to have children

注意：進程之間是互相獨立的，主進程代碼運行結束，守護進程隨即終止。

1、守護進程的啟動

import os
import time
from multiprocessing import Process

class Myprocess(Process):
    def __init__(self,person):
        super().__init__()
        self.person = person
    def run(self):
        print(os.getpid(),self.name)
        print('%s正在和女主播聊天' %self.person)


p=Myprocess('哪吒')
p.daemon=True # 一定要在p.start()前設置,設置p為守護進程,禁止p創建子進程,並且父進程代碼執行結束,p即終止運行
p.start()
time.sleep(10)  # 在sleep時查看進程id對應的進程ps -ef|grep id
print('主')

2、主進程代碼執行結束守護進程立即結束

from multiprocessing import Process

def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")


p1=Process(target=foo)
p2=Process(target=bar)

p1.daemon=True
p1.start()
p2.start()
time.sleep(0.1)
print("main-------")  # 打印該行則主進程代碼結束,則守護進程p1應該被終止.#可能會有p1任務執行的打印信息123,因為主進程打印main----時,p1也執行了,但是隨即被終止，p2可以打印出來。

4、socket聊天並發實例

from socket import *
from multiprocessing import Process

server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5)

def talk(conn,client_addr):
    while True:
        try:
            msg=conn.recv(1024)
            if not msg:break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == '__main__': # windows下start進程一定要寫到這下面
    while True:
        conn,client_addr=server.accept()
        p=Process(target=talk,args=(conn,client_addr))

5、進程對象的其他方法：terminate和is_alive

from multiprocessing import Process
import time
import random

class Myprocess(Process):
    def __init__(self,person):
        self.name=person
        super().__init__()

    def run(self):
        print('%s正在和網紅臉聊天' %self.name)
        time.sleep(random.randrange(1,5))
        print('%s還在和網紅臉聊天' %self.name)


p1=Myprocess('哪吒')
p1.start()

p1.terminate()#關閉進程,不會立即關閉,所以is_alive立刻查看的結果可能還是存活
print(p1.is_alive()) #結果為True

print('開始')
print(p1.is_alive()) #結果為False

6、進程對象的其他屬性：pid和name

class Myprocess(Process):
    def __init__(self,person):
        self.name=person   # name屬性是Process中的屬性，標示進程的名字
        super().__init__()  # 執行父類的初始化方法會覆蓋name屬性
        # self.name = person  # 在這里設置就可以修改進程名字了
        # self.person = person  # 如果不想覆蓋進程名，就修改屬性名稱就可以了
    def run(self):
        print('%s正在和網紅臉聊天' %self.name)
        # print('%s正在和網紅臉聊天' %self.person)
        time.sleep(random.randrange(1,5))
        print('%s正在和網紅臉聊天' %self.name)
        # print('%s正在和網紅臉聊天' %self.person)


p1=Myprocess('哪吒')
p1.start()
print(p1.pid)    #可以查看子進程的進程id

7、參考：

Process([group [, target [, name [, args [, kwargs]]]]])，由該類實例化得到的對象，表示一個子進程中的任務（尚未啟動）

強調：
1.需要使用關鍵字的方式來指定參數
2.args指定的為傳給target函數的位置參數，是一個元組形式，必須有逗號

參數介紹：
•group參數未使用，值始終為None
•target表示調用對象，即子進程要執行的任務
•args表示調用對象的位置參數元組，args=(1,2,'egon',)
•kwargs表示調用對象的字典，kwargs={'name':'egon','age':18}
•name為子進程的名稱

1、方法介紹
•p.start()：啟動進程，並調用該子進程中的p.run()
•p.run()：進程啟動時運行的方法，正是它去調用target指定的函數，我們自定義類的類中一定要實現該方法
•p.terminate()：強制終止進程p，不會進行任何清理操作，如果p創建了子進程，該子進程就成了僵屍進程，使用該方法需要特別小心這種情況。如果p還保存了一個鎖那么也將不會被釋放，進而導致死鎖
•p.is_alive()：如果p仍然運行，返回True
•p.join([timeout])：主線程等待p終止（強調：是主線程處於等的狀態，而p是處於運行的狀態）。timeout是可選的超時時間，需要強調的是，p.join只能join住start開啟的進程，而不能join住run開啟的進程

2、屬性介紹
•p.daemon：默認值為False，如果設為True，代表p為后台運行的守護進程，當p的父進程終止時，p也隨之終止，並且設定為True后，p不能創建自己的新進程，必須在p.start()之前設置
•p.name：進程的名稱
•p.pid：進程的pid
•p.exitcode：進程在運行時為None、如果為–N，表示被信號N結束(了解即可)
•p.authkey：進程的身份驗證鍵,默認是由os.urandom()隨機生成的32字符的字符串。這個鍵的用途是為涉及網絡連接的底層進程間通信提供安全性，這類連接只有在具有相同的身份驗證鍵時才能成功（了解即可）

二、進程同步(multiprocess.Lock)

當多個進程使用同一份數據資源的時候，就會引發數據安全或順序混亂問題。

1、多進程搶占輸出資源

import os
import time
import random
from multiprocessing import Process


def work(n):
    print('%s: %s is running' % (n, os.getpid()))
    time.sleep(random.random())
    print('%s:%s is done' % (n, os.getpid()))


if __name__ == '__main__':
    for i in range(3):
        p = Process(target=work, args=(i,))
        p.start()

# 0: 15620 is running
# 1: 19688 is running
# 2: 15892 is running
# 1:19688 is done
# 0:15620 is done
# 2:15892 is done

2、使用鎖維護執行順序

由並發變成了串行,犧牲了運行效率,但避免了競爭，確實會浪費了時間，卻保證了數據的安全。

import os
import time
import random
from multiprocessing import Process,Lock

def work(lock,n):
    lock.acquire()
    print('%s: %s is running' % (n, os.getpid()))
    time.sleep(random.random())
    print('%s: %s is done' % (n, os.getpid()))
    lock.release()

if __name__ == '__main__':
    lock=Lock()
    for i in range(3):
        p=Process(target=work,args=(lock,i))
        p.start()
        
# 1: 24776 is running
# 1: 24776 is done
# 0: 23588 is running
# 0: 23588 is done
# 2: 27308 is running
# 2: 27308 is done

3、多進程同時搶購余票

# 文件db的內容為：{"count":5}
# 注意一定要用雙引號，不然json無法識別
# 並發運行，效率高，但競爭寫同一文件，數據寫入錯亂
from multiprocessing import Process, Lock
import time, json, random


def search():
    dic = json.load(open('db'))
    print('剩余票數%s' % dic['count'])


def get():
    dic = json.load(open('db'))
    time.sleep(random.random())  # 模擬讀數據的網絡延遲
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(random.random())  # 模擬寫數據的網絡延遲
        json.dump(dic, open('db', 'w'))
        print('購票成功')
    else:
        print('購票失敗')


def task(lock):
    search()
    lock.acquire()
    get()
    lock.release()


if __name__ == '__main__':
    lock = Lock()
    for i in range(10):  # 模擬並發10個客戶端搶票
        p = Process(target=task, args=(lock,))
        p.start()

雖然可以用文件共享數據實現進程間通信，但問題是：

效率低（共享數據基於文件，而文件是硬盤上的數據）
需要自己加鎖處理

因此我們最好找尋一種解決方案能夠兼顧：

效率高（多個進程共享一塊內存的數據）
幫我們處理好鎖問題。這就是mutiprocessing模塊為我們提供的基於消息的IPC通信機制：隊列和管道。

隊列和管道都是將數據存放於內存中，隊列又是基於（管道+鎖）實現的，可以讓我們從復雜的鎖問題中解脫出來，我們應該盡量避免使用共享數據，盡可能使用消息傳遞和隊列，避免處理復雜的同步和鎖問題，而且在進程數目增多時，往往可以獲得更好的可獲展性。

三、進程間通信IPC(Inter-Process Communication) (multiprocess.Queue)

1、概念介紹——隊列multiprocess.Queue

創建共享的進程隊列，Queue是多進程安全的隊列，可以使用Queue實現多進程之間的數據傳遞。

Queue([maxsize])創建共享的進程隊列。
參數：maxsize是隊列中允許的最大項數。如果省略此參數，則無大小限制。

底層隊列使用管道和鎖定實現。另外，還需要運行支持線程以便隊列中的數據傳輸到底層管道中。

2、方法介紹

q.get( [ block [ ,timeout ] ] )：返回q中的一個項目。如果q為空，此方法將阻塞，直到隊列中有項目可用為止。block用於控制阻塞行為，默認為True. 如果設置為False，將引發Queue.Empty異常（定義在Queue模塊中）。timeout是可選超時時間，用在阻塞模式中。如果在制定的時間間隔內沒有項目變為可用，將引發Queue.Empty異常。
q.get_nowait() ：同q.get(False)方法。
q.put(item [, block [,timeout ] ] ) ：將item放入隊列。如果隊列已滿，此方法將阻塞至有空間可用為止。block控制阻塞行為，默認為True。如果設置為False，將引發Queue.Empty異常（定義在Queue庫模塊中）。timeout指定在阻塞模式中等待可用空間的時間長短。超時后將引發Queue.Full異常。
q.qsize() ：返回隊列中目前項目的正確數量。此函數的結果並不可靠，因為在返回結果和在稍后程序中使用結果之間，隊列中可能添加或刪除了項目。在某些系統上，此方法可能引發NotImplementedError異常。
q.empty() ：如果調用此方法時 q為空，返回True。如果其他進程或線程正在往隊列中添加項目，結果是不可靠的。也就是說，在返回和使用結果之間，隊列中可能已經加入新的項目。
q.full() ：如果q已滿，返回為True. 由於線程的存在，結果也可能是不可靠的（參考q.empty()方法）。
q.close() ：關閉隊列，防止隊列中加入更多數據。調用此方法時，后台線程將繼續寫入那些已入隊列但尚未寫入的數據，但將在此方法完成時馬上關閉。如果q被垃圾收集，將自動調用此方法。關閉隊列不會在隊列使用者中生成任何類型的數據結束信號或異常。例如，如果某個使用者正被阻塞在get()操作上，關閉生產者中的隊列不會導致get()方法返回錯誤。
q.cancel_join_thread() ：不會再進程退出時自動連接后台線程。這可以防止join_thread()方法阻塞。
q.join_thread() ：連接隊列的后台線程。此方法用於在調用q.close()方法后，等待所有隊列項被消耗。默認情況下，此方法由不是q的原始創建者的所有進程調用。調用q.cancel_join_thread()方法可以禁止這種行為。

3、代碼實例——multiprocess.Queue

1、單看隊列用法

這個例子還沒有加入進程通信，只是先來看看隊列為我們提供的方法，以及這些方法的使用和現象。

'''
multiprocessing模塊支持進程間通信的兩種主要形式:管道和隊列
都是基於消息傳遞實現的,但是隊列接口
'''

from multiprocessing import Queue

q = Queue(3)

# put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(2)
q.put(1)
# q.put(3)   # 如果隊列已經滿了，程序就會停在這里，等待數據被別人取走，再將數據放入隊列。
# 如果隊列中的數據一直不被取走，程序就會永遠停在這里。
try:
    q.put_nowait(3)  # 可以使用put_nowait，如果隊列滿了不會阻塞，但是會因為隊列滿了而報錯。
except:  # 因此我們可以用一個try語句來處理這個錯誤。這樣程序不會一直阻塞下去，但是會丟掉這個消息。
    print('隊列已經滿了')

# 因此，我們再放入數據之前，可以先看一下隊列的狀態，如果已經滿了，就不繼續put了。
print(q.full())  # True

print(q.get())  # 3
print(q.get())  # 2
print(q.get())  # 1
# print(q.get()) # 同put方法一樣，如果隊列已經空了，那么繼續取就會出現阻塞。
try:
    q.get_nowait(3)  # 可以使用get_nowait，如果隊列滿了不會阻塞，但是會因為沒取到值而報錯。
except:  # 因此我們可以用一個try語句來處理這個錯誤。這樣程序不會一直阻塞下去。
    print('隊列已經空了')

print(q.empty())  # True,空了

2、子進程發送數據給父進程

一個queue的簡單應用，使用隊列q對象調用get函數來取得隊列中最先進入的數據。

import time
from multiprocessing import Process, Queue


def f(q):
    q.put([time.asctime(), 'from Eva', 'hello'])  # 調用主函數中p進程傳遞過來的進程參數 put函數為向隊列中添加一條數據。


if __name__ == '__main__':
    q = Queue()  # 創建一個Queue對象
    p = Process(target=f, args=(q,))  # 創建一個進程
    p.start()
    print(q.get())
    p.join()

# ['Mon Dec  9 18:27:08 2019', 'from Eva', 'hello']

3、批量生產數據放入隊列再批量獲取結果

import os
import time
import multiprocessing


# 向queue中輸入數據的函數
def inputQ(queue):
    info = str(os.getpid()) + '(put):' + str(time.asctime())
    queue.put(info)


# 向queue中輸出數據的函數
def outputQ(queue):
    info= queue.get()
    print('%s%s%s' % (str(os.getpid()), '(get):', info))


# Main
if __name__ == '__main__':
    multiprocessing.freeze_support()
    record1 = []  # store input processes
    record2 = []  # store output processes
   
 queue = multiprocessing.Queue(3)

    # 輸入進程
    for i in range(10):
        process = multiprocessing.Process(target=inputQ, args=(queue,))
        process.start()
        record1.append(process)

    # 輸出進程
    for i in range(10):
        process = multiprocessing.Process(target=outputQ, args=(queue,))
        process.start()
        record2.append(process)

    for p in record1:
        p.join()

    for p in record2:
        p.join()

# 17568(get):3208(put):Mon Dec  9 18:29:17 2019
# 27620(get):24024(put):Mon Dec  9 18:29:17 2019
# 19780(get):21716(put):Mon Dec  9 18:29:17 2019
# 27576(get):27608(put):Mon Dec  9 18:29:17 2019
# 11304(get):10668(put):Mon Dec  9 18:29:18 2019
# 19732(get):20548(put):Mon Dec  9 18:29:18 2019
# 18120(get):25360(put):Mon Dec  9 18:29:18 2019
# 24752(get):21764(put):Mon Dec  9 18:29:18 2019
# 19848(get):7604(put):Mon Dec  9 18:29:18 2019
# 13888(get):10376(put):Mon Dec  9 18:29:18 2019

4、生產者消費者模型

在並發編程中使用生產者和消費者模式能夠解決絕大多數並發問題。該模式通過平衡生產線程和消費線程的工作能力來提高程序的整體處理數據的速度。

1、為什么要使用生產者和消費者模式

在線程世界里，生產者就是生產數據的線程，消費者就是消費數據的線程。在多線程開發當中，如果生產者處理速度很快，而消費者處理速度很慢，那么生產者就必須等待消費者處理完，才能繼續生產數據。同樣的道理，如果消費者的處理能力大於生產者，那么消費者就必須等待生產者。為了解決這個問題於是引入了生產者和消費者模式。

2、什么是生產者消費者模式

生產者消費者模式是通過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，所以生產者生產完數據之后不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列里取，阻塞隊列就相當於一個緩沖區，平衡了生產者和消費者的處理能力。

3、基於隊列實現生產者消費者模型

import os
import random
import time
from multiprocessing import Process, Queue


def producer(q):
    for i in range(10):
        time.sleep(random.randint(1, 3))
        res = '包子%s' % i
        q.put(res)
        print('%s 生產了 %s' % (os.getpid(), res))


def consumer(q):
    while True:
        res = q.get()
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生產者們:即廚師們
    p1 = Process(target=producer, args=(q,))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))

    # 開始
    p1.start()
    c1.start()
    print('主')

此時的問題是主進程永遠不會結束，原因是：生產者p在生產完后就結束了，但是消費者c在取空了q之后，則一直處於死循環中且卡在q.get()這一步。

解決方式無非是讓生產者在生產完畢后，往隊列中再發一個結束信號，這樣消費者在接收到結束信號后就可以break出死循環。

4、改良版——生產者消費者模型

注意：結束信號None，不一定要由生產者發，主進程里同樣可以發，但主進程需要等生產者結束后才應該發送該信號。

from multiprocessing import Process, Queue
import time, random, os


def producer(q):
    for i in range(10):
        time.sleep(random.randint(1, 3))
        res = '包子%s' % i
        q.put(res)
        print('%s 生產了 %s' % (os.getpid(), res))
    q.put(None)# 發送結束信號

def consumer(q):
    while True:
        res = q.get()
        if res is None: break  # 收到結束信號則結束
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生產者們:即廚師們
    p1 = Process(target=producer, args=(q,))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))

    # 開始
    p1.start()
    c1.start()
    print('主')

5、主進程在生產者生產完畢后發送結束信號None

from multiprocessing import Process, Queue
import time, random, os


def producer(q):
    for i in range(2):
        time.sleep(random.randint(1, 3))
        res = '包子%s' % i
        q.put(res)
        print('%s 生產了 %s' % (os.getpid(), res))


def consumer(q):
    while True:
        res = q.get()
        if res is None: break  # 收到結束信號則結束
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生產者們:即廚師們
    p1 = Process(target=producer, args=(q,))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))

    # 開始
    p1.start()
    c1.start()

 p1.join() q.put(None)# 發送結束信號

    print('主')

但上述解決方式，在有多個生產者和多個消費者時，我們則需要用一個很low的方式去解決

6、多個消費者的例子：有幾個消費者就需要發送幾次結束信號

from multiprocessing import Process, Queue
import time, random, os


def producer(name, q):
    for i in range(2):
        time.sleep(random.randint(1, 3))
        res = '%s%s' % (name, i)
        q.put(res)
        print('%s 生產了 %s' % (os.getpid(), res))


def consumer(q):
    while True:
        res = q.get()
        if res is None: break  # 收到結束信號則結束
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生產者們:即廚師們
    p1 = Process(target=producer, args=('包子', q))
    p2 = Process(target=producer, args=('骨頭', q))
    p3 = Process(target=producer, args=('泔水', q))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))
    c2 = Process(target=consumer, args=(q,))

    # 開始
    p1.start()
    p2.start()
    p3.start()
    c1.start()

    p1.join()  # 必須保證生產者全部生產完畢,才應該發送結束信號
    p2.join() 
    p3.join()
    q.put(None) # 有幾個消費者就應該發送幾次結束信號None 
    q.put(None) # 發送結束信號     
    print('主')

5、JoinableQueue([maxsize])可連接的共享進程隊列

創建可連接的共享進程隊列。這就像是一個Queue對象，但隊列允許項目的使用者通知生產者項目已經被成功處理。通知進程是使用共享的信號和條件變量來實現的。

1、方法介紹

JoinableQueue的實例p除了與Queue對象相同的方法之外，還具有以下方法：

q.task_done()：使用者使用此方法發出信號，表示q.get()返回的項目已經被處理。如果調用此方法的次數大於從隊列中刪除的項目數量，將引發ValueError異常。
q.join()：生產者將使用此方法進行阻塞，直到隊列中所有項目均被處理。阻塞將持續到為隊列中的每個項目均調用q.task_done()方法為止。
下面的例子說明如何建立永遠運行的進程，使用和處理隊列上的項目。生產者將項目放入隊列，並等待它們被處理。

2、 JoinableQueue隊列實現消費之生產者模型

from multiprocessing import Process, JoinableQueue
import time, random, os


def producer(name, q):
    for i in range(10):
        time.sleep(random.randint(1, 3))
        res = '%s%s' % (name, i)
        q.put(res)
        print('%s 生產了 %s' % (os.getpid(), res))
    q.join()  # 生產完畢，使用此方法進行阻塞，直到隊列中所有項目均被處理。


def consumer(q):
    while True:
        res = q.get()
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))
        q.task_done()  # 向q.join()發送一次信號,證明一個數據已經被取走了


if __name__ == '__main__':
    q = JoinableQueue()
    # 生產者們:即廚師們
    p1 = Process(target=producer, args=('包子', q))
    p2 = Process(target=producer, args=('骨頭', q))
    p3 = Process(target=producer, args=('泔水', q))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))
    c2 = Process(target=consumer, args=(q,))
    c1.daemon = True
    c2.daemon = True

    # 開始
    p_l = [p1, p2, p3, c1, c2]
    for p in p_l:
        p.start()

    p1.join()
    p2.join()
    p3.join()
    print('主')

    # 主進程等--->p1,p2,p3等---->c1,c2
    # p1,p2,p3結束了,證明c1,c2肯定全都收完了p1,p2,p3發到隊列的數據
    # 因而c1,c2也沒有存在的價值了,不需要繼續阻塞在進程中影響主進程了。應該隨着主進程的結束而結束,所以設置成守護進程就可以了。

四、進程池(multiprocess.Pool）

1、概念介紹——multiprocess.Pool

Pool([numprocess [,initializer [, initargs]]])：創建進程池

numprocess：要創建的進程數，如果省略，將默認使用cpu_count()的值
initializer：是每個工作進程啟動時要執行的可調用對象，默認為None
initargs：是要傳給initializer的參數組

2、主要方法

p.apply(func [, args [, kwargs]])：在一個池工作進程中執行func(*args,**kwargs),然后返回結果。需要強調的是：此操作並不會在所有池工作進程中並執行func函數。如果要通過不同參數並發地執行func函數，必須從不同線程調用p.apply()函數或者使用p.apply_async()
p.apply_async(func [, args [, kwargs]])：在一個池工作進程中執行func(*args,**kwargs),然后返回結果。此方法的結果是AsyncResult類的實例，callback是可調用對象，接收輸入參數。當func的結果變為可用時，將理解傳遞給callback。callback禁止執行任何阻塞操作，否則將接收其他異步操作中的結果。
p.close()：關閉進程池，防止進一步操作。如果所有操作持續掛起，它們將在工作進程終止前完成
P.join()：等待所有工作進程退出。此方法只能在close()或teminate()之后調用
obj.get()：返回結果，如果有必要則等待結果到達。timeout是可選的。如果在指定時間內還沒有到達，將引發一場。如果遠程操作中引發了異常，它將在調用此方法時再次被引發。
obj.ready()：如果調用完成，返回True
obj.successful()：如果調用完成且沒有引發異常，返回True，如果在結果就緒之前調用此方法，引發異常
obj.wait([timeout])：等待結果變為可用。
obj.terminate()：立即終止所有工作進程，同時不執行任何清理或結束任何掛起工作。如果p被垃圾回收，將自動調用此函數。

3、代碼實例——multiprocess.Pool

1、同步

import os,time
from multiprocessing import Pool

def work(n):
    print('%s run' %os.getpid())
    time.sleep(3)
    return n**2

if __name__ == '__main__':
    p=Pool(3)

 #進程池中從無到有創建三個進程,以后一直是這三個進程在執行任務
    res_l=[]
    for i in range(10):
        res=p.apply(work,args=(i,))

 # 同步調用，直到本次任務執行完畢拿到res，等待任務work執行的過程中可能有阻塞也可能沒有阻塞
                                    # 但不管該任務是否存在阻塞，同步調用都會在原地等着
    print(res_l)

2、異步

import os
import time
import random
from multiprocessing import Pool


def work(n):
    print('%s run' % os.getpid())
    time.sleep(random.random())
    return n ** 2


if __name__ == '__main__':
    p = Pool(3)  # 進程池中從無到有創建三個進程,以后一直是這三個進程在執行任務
    res_l = []
    for i in range(10):
        res= p.apply_async(work, args=(i,))

  # 異步運行，根據進程池中有的進程數，每次最多3個子進程在異步執行
        # 返回結果之后，將結果放入列表，歸還進程，之后再執行新的任務
        # 需要注意的是，進程池中的三個進程不會同時開啟或者同時結束
        # 而是執行完一個就釋放一個進程，這個進程就去接收新的任務。
        res_l.append(res)

    # 異步apply_async用法：如果使用異步提交的任務，主進程需要使用join，等待進程池內任務都處理完，然后可以用get收集結果
    # 否則，主進程結束，進程池可能還沒來得及執行，也就跟着一起結束了
    p.close()
    p.join()
    for res in res_l:
        print(res.get()

 )  # 使用get來獲取apply_aync的結果,如果是apply,則沒有get方法,因為apply是同步執行,立刻獲取結果,也根本無需get

4、進程池版socket並發聊天練習

1、 server

#Pool內的進程數默認是cpu核數，假設為4（查看方法os.cpu_count()）
#開啟6個客戶端，會發現2個客戶端處於等待狀態
#在每個進程內查看pid，會發現pid使用為4個，即多個客戶端公用4個進程
from socket import *
from multiprocessing import Pool
import os

server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5)

def talk(conn):
    print('進程pid: %s' %os.getpid())
    while True:
        try:
            msg=conn.recv(1024)
            if not msg:break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == '__main__':
    p=Pool(4)
    while True:
        conn,*_=server.accept()
        p.apply_async(talk,args=(conn,))
        # p.apply(talk,args=(conn,client_addr)) #同步的話，則同一時間只有一個客戶端能訪問

2、 client

from socket import *

client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))


while True:
    msg=input('>>: ').strip()
    if not msg:continue

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

發現：並發開啟多個客戶端，服務端同一時間只有4個不同的pid，只能結束一個客戶端，另外一個客戶端才會進來。

5、回調函數

需要回調函數的場景：進程池中任何一個任務一旦處理完了，就立即告知主進程：我好了額，你可以處理我的結果了。主進程則調用一個函數去處理該結果，該函數即回調函數

我們可以把耗時間（阻塞）的任務放到進程池中，然后指定回調函數（主進程負責執行），這樣主進程在執行回調函數時就省去了I/O的過程，直接拿到的是任務的結果。

1、使用多進程請求多個url來減少網絡等待浪費的時間

from multiprocessing import Pool
import requests
import json
import os


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


def pasrse_page(res):
    print('<進程%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 = Pool(3)
    res_l = []
    for url in urls:
        res =p.apply_async(get_page, args=(url,), callback=pasrse_page)
        res_l.append(res)

    p.close()
    p.join()
    print([res.get() for res in res_l])  # 拿到的是get_page的結果,其實完全沒必要拿該結果,該結果已經傳給回調函數處理了

'''
打印結果:
<進程3388> get https://www.baidu.com
<進程3389> get https://www.python.org
<進程3390> get https://www.openstack.org
<進程3388> get https://help.github.com/
<進程3387> parse https://www.baidu.com
<進程3389> get http://www.sina.com.cn/
<進程3387> parse https://www.python.org
<進程3387> parse https://help.github.com/
<進程3387> parse http://www.sina.com.cn/
<進程3387> parse https://www.openstack.org
[{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html>\r\n...',...}]
'''

2、爬蟲實例

import re
from urllib.request import urlopen
from multiprocessing import Pool

def get_page(url,pattern):
    response=urlopen(url).read().decode('utf-8')
    return pattern,response

def parse_page(info):
    pattern,page_content=info
    res=re.findall(pattern,page_content)
    for item in res:
        dic={
            'index':item[0].strip(),
            'title':item[1].strip(),
            'actor':item[2].strip(),
            'time':item[3].strip(),
        }
        print(dic)
if __name__ == '__main__':
    regex = r'<dd>.*?<.*?class="board-index.*?>(\d+)</i>.*?title="(.*?)".*?class="movie-item-info".*?<p class="star">(.*?)</p>.*?<p class="releasetime">(.*?)</p>'
    pattern1=re.compile(regex,re.S)

    url_dic={
        'http://maoyan.com/board/7':pattern1,
    }

    p=Pool()
    res_l=[]
    for url,pattern in url_dic.items():
        res=p.apply_async(get_page,args=(url,pattern),
         callback=parse_page) res_l.append(res) for i in res_l: i.get()

6、無需回調函數

如果在主進程中等待進程池中所有任務都執行完畢后，再統一處理結果，則無需回調函數。

from multiprocessing import Pool
import time,random,os

def work(n):
    time.sleep(1)
    return n**2
if __name__ == '__main__':
    p=Pool()

    res_l=[]
    for i in range(10):
        res=p.apply_async(work,args=(i,))
        res_l.append(res)

    p.close()
    p.join() #等待進程池中所有進程執行完畢

    nums=[]
    for res in res_l:
        nums.append(res.get() ) #拿到所有結果
    print(nums) #主進程拿到所有的處理結果,可以在主進程中進行統一進行處理

進程池的其他實現方法：https://docs.python.org/dev/library/concurrent.futures.html

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 Python程序中的進程操作-進程池(multiprocess.Pool) Python程序中的進程操作-進程間通信(multiprocess.Queue) python之進程（multiprocess） Python程序中的進程操作 python程序中的進程操作 python之threading與multiprocess，進程與線程 Python程序中的進程操作-進程間通信多進程操作-進程隊列multiprocess.Queue的使用 multiprocessing.Process() ----------python中的多進程 Python multiprocess模塊(下)