multiprocessor（下)

 

1、數據共享

展望將來，基於消息傳遞的併發編程是大勢所趨即使是使用線程，推薦作法也是將程序設計爲大量獨立的線程集合，經過消息隊列交換數據。這樣極大地減小了對使用鎖定和其餘同步手段的需求，還能夠擴展到分佈式系統中。

進程間應該儘可能避免通訊，即使須要通訊，也應該選擇進程安全的工具來避免加鎖帶來的問題，應該儘可能避免使用本節所講的共享數據的方式，之後咱們會嘗試使用數據庫來解決進程之間的數據共享問題。

進程之間數據共享的模塊之一Manager模塊。

 

 

 

​x

 

 

 

 

進程間數據是獨立的，能夠藉助於隊列或管道實現通訊，兩者都是基於消息傳遞的

雖然進程間數據獨立，但能夠經過Manager實現數據共享，事實上Manager的功能遠不止於此

​

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

​

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array.

 

 

多進程共同去處理共享數據的時候，就和咱們多進程同時去操做一個文件中的數據是同樣的，不加鎖就會出現錯誤的結果，進程不安全的，因此也須要加鎖。

 

 

 

xxxxxxxxxx

 

 

 

 

from multiprocessing import Manager,Process,Lock

def work(d,lock):

    with lock: #不加鎖而操做共享的數據,確定會出現數據錯亂

        d['count']-=1

​

if __name__ == '__main__':

    lock=Lock()

    with Manager() as m:

        dic=m.dict({'count':100})

        p_l=[]

        for i in range(100):

            p=Process(target=work,args=(dic,lock))

            p_l.append(p)

            p.start()

        for p in p_l:

            p.join()

        print(dic)   #0,加了鎖後

#100減100次1這麼慢? 不是減操做形成的 而是開啓進程 管理進程 銷燬進程拖慢了程序的執行速度

# 爲何在這裏出現了數據不安全的現象?

# 什麼狀況下會出現數據不安全 : Manager類當中對字典\列表 += -= *= /=

# 如何解決 : 加鎖        

 

 

• 總結一下：進程之間通訊：隊列、管道、數據共享也算

  • 下面要講的信號量和事件也至關於鎖，也是全局的，全部進程都能拿到這些鎖的狀態，進程之間這些鎖啊信號量啊事件啊等等的通訊，其實底層仍是socekt，只不過是基於文件的socket通訊，而不是跟上面的數據共享啊空間共享啊之類的機制，咱們以前學的是基於網絡的socket通訊，還記得socket的兩個家族嗎，一個文件的一個網絡的，因此未來若是說這些鎖之類的報錯，可能你看到的就是相似於socket的錯誤，簡單知道一下就能夠啦~~~

    工做中經常使用的是鎖，信號量和事件不經常使用，可是信號量和事件面試的時候會問到，你能知道就行啦

2、信息量（進程中的，也有線程的）

• 信息量Semaphore介紹

  •  

     

     

    xxxxxxxxxx

     

     

     

     

    互斥鎖同時只容許一個線程更改數據，而信號量Semaphore是同時容許必定數量的線程更改數據 。

    假設商場裏有4個迷你唱吧，因此同時能夠進去4我的，若是來了第五我的就要在外面等待，等到有人出來才能再進去玩。

    實現：

    信號量同步基於內部計數器，每調用一次acquire()，計數器減1；每調用一次release()，計數器加1.當計數器爲0時，acquire()調用被阻塞。這是迪科斯徹（Dijkstra）信號量概念P()和V()的Python實現。信號量同步機制適用於訪問像服務器這樣的有限資源。

    信號量與進程池的概念很像，可是要區分開，信號量涉及到加鎖的概念

     

     

  • 好比大保健：提早設定好，一個房間只有4個牀（計數器如今爲4），那麼同時只能四我的進來，誰先來的誰先佔一個牀（acquire，計數器減1），4個牀滿了以後（計數器爲0了），第五我的就要等着，等其中一我的出來（release，計數器加1），他就去佔用那個牀了。

     

     

     

    xxxxxxxxxx

     

     

     

     

    from multiprocessing import Process,Semaphore

    import time,random

    ​

    def go_ktv(sem,user):

        sem.acquire()

        print('%s 佔到一間ktv小屋' %user)

        time.sleep(random.randint(0,3)) #模擬每一個人在ktv中待的時間不一樣

        sem.release()

    ​

    if __name__ == '__main__':

        sem=Semaphore(4)

        p_l=[]

        for i in range(13):

            p=Process(target=go_ktv,args=(sem,'user%s' %i,))

            p.start()

            p_l.append(p)

    ​

        for i in p_l:

            i.join()

        print('============》')

    ​

    #模擬6我的去ktv唱歌

    import time

    import random

    from multiprocessing import Process,Semaphore

    def ktv(i):

        print('person %s 進來唱歌了'%i)

        time.sleep(random.randint(1,5))

        print('person %s 從ktv出去了'%i)

    if __name__ == '__main__':

        for i in range(6):  # 模擬6我的

            Process(target=ktv,args=(i,)).start()  #結果發現有問題，6我的一擁而入，沒有次序亂了

    #使用semaphore來設置一次進去的人數，KTV 4我的

    import time

    import random

    from multiprocessing import Process,Semaphore

    def ktv(i,sem):

        sem.acquire()  #取得鎖

        print('person %s 進來唱歌了'%i)

        time.sleep(random.randint(1,5))

        print('person %s 從ktv出去了'%i)

        sem.release()  #釋放鎖

    if __name__ == '__main__':

        sem = Semaphore(4)  #初始化信號量，數量爲4

        for i in range(6):  # 模擬6我的

            Process(target=ktv,args=(i,sem)).start()

    #在同一時間，最多有4我的進去

    acquire()是一個阻塞行爲，信號量和鎖有點相似

    那麼它們之間的區別在於：

    信號量，至關於計數器

    它是鎖+計數器

    調用acquire() 計數器-1

    當計數器到 0 時，再調用 acquire() 就會阻塞，直到其餘線程來調用release()

    調用release() 計數器+1   

     

     

     

3、事件

 

4、進程池

 

5、死鎖（進程）

• 在多道程序系統中，因爲多個進程的併發執行，改善了系統資源的利用率並提升了系統的處理能力。然而，多個進程的併發執行也帶來了新的問題——死鎖。所謂死鎖是指多個進程因競爭資源而形成的一種僵局，若無外力做用，這些進程都將沒法向前推動。

• 出現的緣由：

• 代碼示例

   

   

   

  xxxxxxxxxx

   

   

   

   

  from threading import Thread,Lock

  import time

  mutexA=Lock()

  mutexB=Lock()

  ​

  class MyThread(Thread):

      def run(self):

          self.func1()

          self.func2()

      def func1(self):

          mutexA.acquire()

          print('\033[41m%s 拿到A鎖\033[0m' %self.name)

  ​

          mutexB.acquire()

          print('\033[42m%s 拿到B鎖\033[0m' %self.name)

          mutexB.release()

  ​

          mutexA.release()

  ​

      def func2(self):

          mutexB.acquire()

          print('\033[43m%s 拿到B鎖\033[0m' %self.name)

          time.sleep(2)

  ​

          mutexA.acquire()

          print('\033[44m%s 拿到A鎖\033[0m' %self.name)

          mutexA.release()

  ​

          mutexB.release()

  ​

  if __name__ == '__main__':

      for i in range(5):

          t=MyThread()

          t.start()