并发编程4

Posted 2020-11-01 木夂各

守护进程 　　

　　语法：obj.daemon = True 　　　　必须放在obj.start()之前

　　特点：守护进程内无法在开启子进程，否则抛出异常。AssertionError: daemonic processes are not allowed to have children
　　　　   父进程代码执行结束守护进程也必须结束

from multiprocessing import Process
import time


def task(name):
    print(‘%s is running‘ % name)
    time.sleep(3)

if __name__ == ‘__main__‘:
    obj = Process(target=task, args=(‘egon‘,))
    obj.daemon=True
    obj.start()  # 发送信号给操作系统
    print(‘主‘)

 

互斥锁

　　强调：必须lock.acquire()一次，然后lock.release()释放一次，才能继续lock.acquire()，不能连续lock.acquire()

互斥锁和join的区别:
　　大前提：二者的原理都是一样，都是将并发变成串行，从而保证有序，

　　区别一：join是按照人为指定的顺序执行，而互斥锁只所有进程平等竞争，谁先抢到谁先执行

　　

from multiprocessing import Process,Lock
import time,random

mutex=Lock()

def task1(lock):
    lock.acquire() #
    print(‘task1:名字是egon‘)
    time.sleep(random.randint(1,3))
    print(‘task1:性别是male‘)
    time.sleep(random.randint(1,3))
    print(‘task1:年龄是18‘)
    lock.release()

def task2(lock):
    lock.acquire()
    print(‘task2:名字是alex‘)
    time.sleep(random.randint(1,3))
    print(‘task2:性别是male‘)
    time.sleep(random.randint(1,3))
    print(‘task2:年龄是78‘)
    lock.release()


def task3(lock):
    lock.acquire()
    print(‘task3:名字是lxx‘)
    time.sleep(random.randint(1,3))
    print(‘task3:性别是female‘)
    time.sleep(random.randint(1,3))
    print(‘task3:年龄是30‘)
    lock.release()


if __name__ == ‘__main__‘:
    p1=Process(target=task1,args=(mutex,))
    p2=Process(target=task2,args=(mutex,))
    p3=Process(target=task3,args=(mutex,))

    # p1.start()
    # p1.join()
    # p2.start()
    # p2.join()
    # p3.start()
    # p3.join()

    p1.start()
    p2.start()
    p3.start()

示例代码

 

　　区别二：互斥锁可以让一部分代码(修改共享数据的代码)串行，而join只能让代码整体串行

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

mutex=Lock()

def search():
    time.sleep(random.randint(1,3))
    with open(‘db.json‘,‘r‘,encoding=‘utf-8‘) as f:
        dic=json.load(f)
        print(‘%s 剩余票数:%s‘ %(os.getpid(),dic[‘count‘]))

def get():
    with open(‘db.json‘,‘r‘,encoding=‘utf-8‘) as f:
        dic=json.load(f)
    if dic[‘count‘] > 0:
        dic[‘count‘]-=1
        time.sleep(random.randint(1,3))
        with open(‘db.json‘,‘w‘,encoding=‘utf-8‘) as f:
            json.dump(dic,f)
        print(‘%s 购票成功‘ %os.getpid())

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

if __name__ == ‘__main__‘:
    for i in range(10):
        p=Process(target=task,args=(mutex,))
        p.start()
        # p.join()

示例代码

 

#加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信，但问题是：
1.效率低（共享数据基于文件，而文件是硬盘上的数据）
2.需要自己加锁处理

 

 

#因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
1 队列和管道都是将数据存放于内存中
2 队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，
我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性

 

IPC通信机制

　进程之间通信必须找到一种介质，该介质必须满足：
　　1，所有进程共享
　　2，必须是内存空间
　　附加：最好能够处理好所的问题

 

# from multiprocessing import Process,Manager,Lock
# import time
#
# mutex=Lock()
#
# def task(dic,lock):
#     lock.acquire()
#     temp=dic[‘num‘]
#     time.sleep(0.1)
#     dic[‘num‘]=temp-1
#     lock.release()
#
# if __name__ == ‘__main__‘:
#     m=Manager()
#     dic=m.dict({‘num‘:10})
#
#     l=[]
#     for i in range(10):
#         p=Process(target=task,args=(dic,mutex))
#         l.append(p)
#         p.start()
#
#     for p in l:
#         p.join()
#     print(dic)


from multiprocessing import Queue

了解--mananger模块--实现共享空间

 

 

 

解决方案：

　　Queue:队列---先进先出

　　1.进程共享  2,内存空间  3.自动处理了锁定问题
　　强调：队列用来存放进程之间沟通的消息，数据量不应该过大

　　　　　maxsize的值超过内存限制就变得毫无意义

 Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。 

maxsize是队列中允许最大项数，省略则无大小限制。

主要方法：

1 q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。
2 q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.
3  
4 q.get_nowait():同q.get(False)
5 q.put_nowait():同q.put(False)
6 
7 q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。
8 q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。
9 q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样

其他方法
1 q.cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞
2 q.close():关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如，如果某个使用者正在被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。
3 q.join_thread()：连接队列的后台线程。此方法用于在调用q.close()方法之后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为

Queue语法

‘‘‘
multiprocessing模块支持进程间通信的两种主要形式:管道和队列
都是基于消息传递实现的,推荐队列接口
‘‘‘

from multiprocessing import Process,Queue
import time
q=Queue(3)


#put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(3)
q.put(3)
print(q.full()) #满了

print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #空了

 

 

生产者消费者模型

　　两类角色：
　　　　生产者：负责造数据的的任务比喻为生产者
　　　　消费者：接收生产者造出的数据来做进一步的处理，该类人物被比喻成消费者

　　实现生产者消费者模型三要素：
　　　　1，生产者
　　　　2，消费者
　　　　3，队列

 

　　什么时候使用生产这消费者模型：
　　　　程序中出现明显的两类任务，一类任务负责生产，一类任务负责处理生产的数据

　　好处：
　　　　1，实现了生产者和消费者解耦合
　　　　2，平衡了生产力和消费力，彼此互不影响，即生产可以不停的生产，消费者可以不停消费，二者互相不沟通，只和队列沟通

from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print(‘\033[45m%s 吃 %s\033[0m‘ %(os.getpid(),res))

def producer(q):
    for i in range(10):
        time.sleep(random.randint(1,3))
        res=‘包子%s‘ %i
        q.put(res)
        print(‘\033[44m%s 生产了 %s\033[0m‘ %(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(‘主‘)

基于队列的生产者消费者模型

#生产者消费者模型总结

    #程序中有两类角色
        一类负责生产数据（生产者）
        一类负责处理数据（消费者）
        
    #引入生产者消费者模型为了解决的问题是：
        平衡生产者与消费者之间的工作能力，从而提高程序整体处理数据的速度
        
    #如何实现：
        生产者<-->队列<——>消费者
    #生产者消费者模型实现类程序的解耦和