鎖和隊列

一、鎖multiprocessing-Lock

鎖的應用場景：當多個進程須要操做同一個文件/數據的時候；

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

爲保證數據的安全性，多進程中只有去操做一些進程之間能夠共享的數據資源的時候才須要進行加鎖；

枷鎖能夠保證多個進程修改同一塊數據時，同一時間只能有一個任務進行修改，即串行的修改，沒錯速度式慢了，但犧牲了速度卻保證了數據的安全；

模擬查票搶票：

 

import json
import time
from multiprocessing import Process,Lock

"""
模擬查票/搶票
查票是不須要排隊的,可使用多進程,每一個人均可以在同一時間查看
買票爲避免多個進程操做同一個文件形成數據不安全,使用鎖來避免多個進程同時修改同一個文件
"""
def search_ticket(name):
    with open('piao',encoding='utf-8') as f:
        dic = json.load(f)
        print('%s查看餘票爲%s張票'%(name,dic['count']))

def buy_ticket(name):
    with open('piao', encoding='utf-8') as f:
        dic = json.load(f)
    time.sleep(2)
    if dic['count'] >= 1:
        print('%s買到票了'%name)
        dic['count'] -= 1
        time.sleep(2)
        with open('piao','w',encoding='utf-8') as f:
            json.dump(dic,f)
    else:
        print('餘票爲0 %s沒有買到票'%name)

def user(name,lock):
    search_ticket(name)
    print('%s在等待' % name)
    # lock.acquire()    # 拿鑰匙取數據
    # buy_ticket(name)
    # lock.release()    # 歸還鑰匙
    with lock:
        print('%s開始執行了' % name)
        buy_ticket(name)

if __name__ == '__main__':
    lock = Lock()  # 建立鎖
    lis = ['alex','wusir','taibai','eva_j']
    for name in lis:
        Process(target=user,args=(name,lock)).start()

模擬查票/搶票

 

#加鎖能夠保證多個進程修改同一塊數據時，同一時間只能有一個任務能夠進行修改，即串行的修改，沒錯，速度是慢了，但犧牲了速度卻保證了數據安全。
雖然能夠用文件共享數據實現進程間通訊，但問題是：
1.效率低（共享數據基於文件，而文件是硬盤上的數據）
2.須要本身加鎖處理

#所以咱們最好找尋一種解決方案可以兼顧：一、效率高（多個進程共享一塊內存的數據）二、幫咱們處理好鎖問題。這就是mutiprocessing模塊爲咱們提供的基於消息的IPC通訊機制：隊列和管道。
隊列和管道都是將數據存放於內存中
隊列又是基於（管道+鎖）實現的，可讓咱們從複雜的鎖問題中解脫出來，
咱們應該儘可能避免使用共享數據，儘量使用消息傳遞和隊列，避免處理複雜的同步和鎖問題，並且在進程數目增多時，每每能夠得到更好的可獲展性。 

二、進程之間的通訊—隊列multiprocessing.Queue

 

進程之間的通訊：IPC(Inter-Process Communication)

multiprocessing.Queue能夠完成進程之間通訊的特殊隊列

from multiprocessing import Queue
def son(q):
    print(q.get())  # 從隊列裏面取值
if __name__ == '__main__':
    q = Queue()   # 建立一個隊列
    Process(target=son,args=(q,)).start()
    q.put('wanhaha')   # 往隊列裏放值

使用隊列完成主進程與子進程的通訊

1、隊列是進程之間通訊的安全保障，自帶了鎖
2、隊列是基於文件家族的socket服務實現的
3、管道Pipe也是基於文件家族的socket服務實現的IPC機制
四、隊列 = 管道 + 鎖
五、管道沒有鎖，不安全

三、生產者消費模型

在併發編程中使用生產者和消費者模式可以解決絕大多數併發問題。該模式經過平衡生產線程和消費線程的工做能力來提升程序的總體處理數據的速度。

爲何要使用生產者和消費者模式

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

什麼是生產者消費者模式

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

import time
from multiprocessing import Queue,Process
def producer(q):
    """
    生產者
    :param q:
    :param name:
    :return:
    """
    for i in range(10):
        time.sleep(0.2)
        q.put('泔水%s'%i)
        print('正在生產泔水%s'%i)
def consumer(q,name):
    """
    消費者
    :param q:
    :return:
    """
    while True:
        s = q.get()
        if not s:break
        time.sleep(0.2)
        print('%s正在吃%s'%(name,s))
if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=producer,args=(q,))
    p1.start()
    c1 = Process(target=consumer,args = (q,'alex'))
    c1.start()
    p1.join()
    q.put(None)

向隊列中添加None來結束消費者中的循環

import time
import random
from multiprocessing import JoinableQueue,Process

def consumer(jq,name):
    while True:
        food = jq.get()
        time.sleep(random.uniform(1,2))
        print('%s吃完%s'%(name,food))
        jq.task_done()

def producer(jq):
    for i in range(10):
        time.sleep(random.random())
        food = '泔水%s'%i
        print('%s生產了%s'%('taibai',food))
        jq.put(food)
    jq.join()

if __name__ == '__main__':
    jq = JoinableQueue(5)
    c1 = Process(target=consumer,args=(jq,'alex'))
    p1 = Process(target=producer,args=(jq,))
    c1.daemon = True  # 守護進程
    c1.start()
    p1.start()
    p1.join()

利用守護進程結束循環

 

邊生產邊吃,供不該求時會get()阻塞住,直到生產者往隊列裏放有值才能繼續吃，每吃一次都會向隊列發送一個信號q.task_done()(至關於告訴隊列我吃完一個了)此時隊列中的計數器會減去一，只要隊列中的計數器不爲0,程序就會被q.join()阻塞住；