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()