Python學習day37-併發編程(3)

時間  2019-11-29
標籤 python 學習 day37 day 併發 編程 欄目 Python 简体版
原文   原文鏈接

 

Python學習day37-併發編程(3)進程鎖LockQueue隊列生產者消費者模型JoinableQueuehtml

Python學習day37-併發編程(3)

進程鎖Lock

上文中咱們提到搶票軟件的原理,其實還能夠作一個小優化,具體就要用到進程鎖這個模塊,進程鎖能夠起到和join相同的做用,可是有區別.node

 
 
  
xxxxxxxxxx
2
 
 
 
 
1 
* join是吧全部的子進程變成了串行
2 
* 進程鎖則是把鎖住的代碼變成了串行,相比之下進程鎖更加方便,實用性泛用性也更強
 
 

示例代碼以下:python

 
 
  
xxxxxxxxxx
43
 
 
 
 
1 
from multiprocessing import Process, Lock
2 
import time
3 
import json
4 
import os
5 
​
6 
​
7 
def search():
8 
    time.sleep(1)  # 模擬網絡io
9 
    with open('db.txt', 'rt', encoding='utf-8')as f:
10 
        res = json.load(f)
11 
        print(f'還剩{res["count"]}張票')
12 
​
13 
​
14 
def get():
15 
    with open('db.txt', 'rt', encoding='utf-8')as f:
16 
        res = json.load(f)
17 
        # print(f'還剩{res["count"]}張票')
18 
    time.sleep(1)  # 模擬網絡io
19 
    if res['count'] > 0:
20 
        res['count'] -= 1
21 
        with open('db.txt', 'wt', encoding='utf-8')as f:
22 
            json.dump(res, f)
23 
            f.flush()
24 
            print(f'{os.getpid()}搶票成功')
25 
            time.sleep(1)  # 模擬網絡io
26 
​
27 
    else:
28 
        print('票已經售空!~!~!~!~!~!~!~!')
29 
​
30 
​
31 
def task(lock):
32 
    # search()
33 
    lock.acquire()  # 進程鎖鎖住,同一時間只能拿有一個用於執行get函數,也就是購買函數
34 
    get()
35 
    lock.release()  # 釋放鎖,下一個用戶能夠操做
36 
​
37 
​
38 
if __name__ == '__main__':
39 
    lock = Lock()  # 寫在main裏面,主進程裏,讓子進程拿到同一把鎖
40 
    for i in range(5):
41 
        p = Process(target=task, args=(lock,))
42 
        p.start()
43 
        # p.join()# 由於加入了進程鎖,因此join用來串行子進程的方式就能夠棄用了
 
 

以上方法雖然相比以前有了必定的改進,但實際上仍是很麻煩,並不適用於併發數量較多的狀況,並且一方面效率比較低,另一方面須要本身加鎖處理.web

因此其實咱們還有一種更好更方便的方式來解決這個問題,就是隊列Queue.編程

Queue隊列

首先咱們瞭解一下進程間通訊的概念便可,簡稱是IPC,全稱是Inter-Process Communication,就是字面意思,進程之間的通訊,咱們一般會採用Queue隊列的方式.json

隊列的概念其實就是管道pipe加進程鎖lock,其做用局勢建立共享的進程隊列,從而來實現進程之間的通訊以及數據傳遞,實際上隊列裏面通常不會放入比較大的數據,大部分時候都是消息或者是一些很是小的數據.windows

Queue(maxsize)就是建立隊列的方法,下面咱們來看一下隊列Queue的部分源碼,筆者已經把那些複雜的實現原理刪掉,咱們只須要了解其有什麼參數以及怎麼使用就好:網絡

 
 
  
xxxxxxxxxx
35
 
 
 
 
1 
class Queue(object):
2 
​
3 
    def __init__(self, maxsize=0, *, ctx):
4 
        pass
5 
    # 這裏maxsize = 0就是隊列的最大長度,若是沒有手動賦值的話隊列就不限長度.
6 
    def put(self, obj, block=True, timeout=None):
7 
pass
8 
    # 這裏,put方法,最經常使用的方法之一,將值放入隊列中
9 
    # 其中的參數,obj即爲要放入的值
10 
    # block是一種阻塞狀態,默認爲True,即阻塞,阻塞的意思就是若是隊列長度已經滿了,put不能放值進去,那麼這個方法就會中止,不會繼續向下執行.
11 
    # timeout即爲時間延時,延時就是put方法在這些時間以內若是不能成功給隊列放入值就會報錯.要注意的是若是block爲False的話延時是沒有做用的,由於此時方法不具備阻塞做用,因此延時天然是無效的.
12 
​
13 
    def get(self, block=True, timeout=None):
14 
        pass
15 
    # get,便是從隊列中取值,其參數和用法和put徹底同樣,只是咱們不用往裏面放值而是按以前put進去的順序往外取值.block和timeout的含義也和上面put同樣
16 
​
17 
    def qsize(self):
18 
        return self._maxsize - self._sem._semlock._get_value()
19 
# 這個方法能夠返回隊列中的值的數量,但並不是可靠值,由於隊列中值會在不斷的變化,使用時候要注意
20 
    def empty(self):
21 
        return not self._poll()
22 
# 判斷隊列是否爲空,返回值爲布爾值True或者False
23 
    def full(self):
24 
        return self._sem._semlock._is_zero()
25 
# 判斷隊列是否已經滿了,返回值爲布爾值True或者False
26 
    def get_nowait(self):
27 
        return self.get(False)
28 
# 用法等同於get,可是沒有block參數,至關於block = False
29 
    def put_nowait(self, obj):
30 
        return self.put(obj, False)
31 
# 用法等同於put,可是沒有block參數,至關於block = False
32 
    def close(self):
33 
pass
34 
    # 字面意思,關閉當前隊列,防止隊列中加入更多的數據
35 
​
 
 

 

生產者消費者模型

  • 生產者:生產數據的任務併發

  • 消費者:處理數據的任務app

    生產者<-->隊列<-->消費者

    生產者能夠不停地生產,能夠達到本身最大的生產效率 消費者能夠不停的消費,也達到了本身最大的消費效率生產者消費者模型: 大大提升了生產者生產的效率和消費者消費的效率

    因此在併發編程中,使用生產者和消費者模型能夠解決絕大多數的併發問題.這個模式經過平衡生產線程和消費線程的工做能力來提升程序的總體處理數據的速度.

    要注意的一點是,隊列Queue並不適合傳輸大型的數據或文件,一般應該傳遞消息之類,上文中也有提到.

    如下爲生產者和消費者的一個示例:

     
     
      
    x
     
     
     
     
    1 
    from multiprocessing import Process, Queue
    2 
    ​
    3 
    ​
    4 
    def producer(q, name, food):
    5 
        '''生產者函數'''
    6 
        for i in range(3):
    7 
            print(f'{name}生產出來{food}{i}')
    8 
            res = f'{food}{i}'
    9 
            q.put(res)
    10 
    ​
    11 
    ​
    12 
    def consumer(q, name):
    13 
        '''消費者函數'''
    14 
        while True:
    15 
            res = q.get(timeout=2)
    16 
            if res is None:# 這裏,當取到隊列中的值是None時,跳出循環,即消費者中止消費
    17 
                print(f'{name}沒東西了,木法吃了')
    18 
                break
    19 
            print(f'{name}吃了{res}')
    20 
    ​
    21 
    ​
    22 
    if __name__ == '__main__':
    23 
        q = Queue()
    24 
        p1 = Process(target=producer, args=(q, 'rocky', '包子'))
    25 
        p2 = Process(target=producer, args=(q, 'tank', '韭菜'))
    26 
        p3 = Process(target=producer, args=(q, 'nick', '蒜泥'))
    27 
        c1 = Process(target=consumer, args=(q, '111'))
    28 
        c2 = Process(target=consumer, args=(q, '222'))
    29 
        p1.start()
    30 
        p2.start()
    31 
        p3.start()
    32 
        c1.start()
    33 
        c2.start()
    34 
        p1.join()
    35 
        p2.join()
    36 
        p3.join()
    37 
        q.put(None)# 爲何這裏要加入兩個None呢,爲了讓兩個消費者都能成功的跳出消費循環,不至於到最後報錯隊列爲空.
    38 
        q.put(None)
    39 
    ​
     
     

    其實這種處理方式有點傻,試想一下,若是這樣處理生產者和消費者之間的關係,那麼每多一個消費者咱們就要多在隊列里加入一個None,這樣會很是的麻煩,並且代碼也會顯得很是的low.

    因此咱們能夠借用JoinableQueue來完成這個事情.

JoinableQueue

JoinableQueue的用法和Queue很是的類似,可是有些許不一樣,讓咱們來看下JoinableQueue的源碼如何:

 
 
  
x
 
 
 
 
1 
class JoinableQueue(Queue):
2 
​
3 
    def __init__(self, maxsize=0, *, ctx):
4 
pass
5 
    # 咱們能夠看到,JoinableQueue的自身init裏面和Queue基本上如出一轍
6 
​
7 
    def put(self, obj, block=True, timeout=None):
8 
pass
9 
    # put的用法和Queue同樣,參數也同樣
10 
​
11 
    def task_done(self):
12 
pass
13 
# 這個就是JoinableQueue和Queue最大的不一樣之一,咱們能夠看到,JoinableQueue裏面並無get方法,因此在調用get方法的時候是調用的JoinableQueue父類,也就是Queue的get方法.
14 
    # 而task_done的意義就是使用者調用此方法的時候會發出一個信號,表示q.get()返回的項目已經被處理掉了.
15 
    def join(self):
16 
pass
17 
    # 這裏的join一樣爲阻塞做用,不一樣的是在JoinableQueue裏面,咱們調用這個方法的時候會阻塞直到隊列裏全部的項目都會處理掉.處理掉的意思就是在全部的q.get()下面都要加上task_done的調用,這樣join纔會再也不阻塞程序的運行.
18 
​
19 
# 上述概念咱們能夠理解爲一個計數器
20 
# 對這個計數器來講:
21 
#   put +1的操做
22 
#   task_done -1的操做
23 
# q.task_done()   完成了一次任務,和get連用,減小計數器的計數
24 
# q.join()   計數器爲零纔會不阻塞
 
 

 

那麼,採用JoinableQueue改造過上面的生產者消費者模型的小實例爲:

 
 
  
x
 
 
1 
from multiprocessing import Process,  JoinableQueue
2 
import time, random
3 
​
4 
​
5 
def producer(q, name, food):
6 
    '''生產者函數'''
7 
    for i in range(3):
8 
        time.sleep(2 * random.random())
9 
        print(f'{name}生產出來{food}{i}')
10 
        res = f'{food}{i}'
11 
        q.put(res)
12 
​
13 
​
14 
def consumer(q, name):
15 
    '''消費者函數'''
16 
    while True:
17 
        res = q.get()
18 
        time.sleep(2 * random.random())
19 
        # if res is None:# 由於斷定方式變了,咱們沒有向隊列里加入None,因此這個判斷無效了
20 
        #     print(f'{name}沒東西了,木法吃了')
21 
        #     break
22 
        print(f'{name}吃了{res}')
23 
        q.task_done()# 這裏是至關於隊列的計數器減一
24 
​
25 
​
26 
if __name__ == '__main__':
27 
    q = JoinableQueue()
28 
    p1 = Process(target=producer, args=(q, 'rocky', '包子'))
29 
    p2 = Process(target=producer, args=(q, 'tank', '韭菜'))
30 
    p3 = Process(target=producer, args=(q, 'nick', '蒜泥'))
31 
    c1 = Process(target=consumer, args=(q, '111'), daemon=True)
32 
    c2 = Process(target=consumer, args=(q, '222'), daemon=True)
33 
    # 這裏在消費者的子進程生成時咱們將其定義爲守護進程,這樣主進程在執行完畢後子進程就會隨之結束,咱們就不用再每添加一個消費者就要往隊列裏添加一個None了.
34 
    p1.start()
 
 
35 
    p2.start()
36 
    p3.start()
37 
    c1.start()
38 
    c2.start()
39 
    p1.join()
40 
    p2.join()
41 
    p3.join()  # 生產者生產完畢
42 
    q.join()  # 這裏是主進程的最後一行代碼
43 
    # 因此把消費者c1,c2作成守護進程的話,主進程最後一行一旦執行經過(q.join()經過也就是隊列爲空),就能夠結束子進程,也就是消費者的兩個子進程就會結束
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程