进程之 Process join方法其他属性与进程Queue

Process join方法 以及其他属性

在主进程运行过程中如果想并发地执行其他的任务，我们可以开启子进程，此时主进程的任务与子进程的任务分两种情况

情况一：在主进程的任务与子进程的任务彼此独立的情况下，主进程的任务先执行完毕后，主进程还需要等待子进程执行完毕，然后统一回收资源。

情况二：如果主进程的任务在执行到某一个阶段时，需要等待子进程执行完毕后才能继续执行，就需要有一种机制能够让主进程检测子进程是否运行完毕，在子进程执行完毕后才继续执行，否则一直在原地阻塞，这就是join方法的作用

join的用法：

from multiprocessing import Process
import time
import random
import os

def task():
    print('%s is piaoing' %os.getpid())
    time.sleep(random.randrange(1,3))
    print('%s is piao end' %os.getpid())

if __name__ == '__main__':
    p=Process(target=task)
    p.start()
    p.join() #等待p停止,才执行下一行代码
    print('主')

会有很多人问有了join是不是程序就会变成穿行了呢，这个你需要明确一点，join这是让主进程等待，子进程是都在运行的，同时启动4个进程，那么四个进程都会同时执行！

from multiprocessing import Process
import time
import random

def task(name):
    print('%s is piaoing' %name)
    time.sleep(random.randint(1,3))
    print('%s is piao end' %name)

if __name__ == '__main__':
    p1=Process(target=task,args=('egon',))
    p2=Process(target=task,args=('alex',))
    p3=Process(target=task,args=('yuanhao',))
    p4=Process(target=task,args=('wupeiqi',))

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

    既然join是等待进程结束, 那么我像下面这样写, 进程不就又变成串行的了吗?
    # 当然不是了, 必须明确：p.join()是让谁等？
    # 很明显p.join()是让主线程等待p的结束，卡住的是主进程而绝非子进程p，
    p1.join()
    p2.join()
    p3.join()
    p4.join()

    print('主')

上面的启动进程可以简化为

p_l=[p1,p2,p3,p4]

for p in p_l:
    p.start()

for p in p_l:
    p.join()

Process对象的其他属性或方法

import multiprocessing
import time


# 开启进程的 第一种方式
def hi(name, x):
    print("start time", time.ctime())
    print("hi %s" % name)
    time.sleep(x)
    print(" %s end time " % name, time.ctime())


if __name__ == '__main__':
    p1 = multiprocessing.Process(target=hi, args=("alex", 3), name="alex Process") #  给进程起个名字
    p2 = multiprocessing.Process(target=hi, args=("egon", 2))
    p1.start()
    print(p1.is_alive())  # 查看进程是否存活
    print(p1.name)  # 查看进程的名字 生成进程的时候可以取名字
    p1.terminate()  # 告诉操作系统 干死p1
    p2.start()
    p1.join()
    p2.join()
    print("主进程")
    print(p1.is_alive())

守护进程

主进程创建子进程，然后将该进程设置成守护自己的进程，守护进程就好比崇祯皇帝身边的老太监，崇祯皇帝已死老太监就跟着殉葬了。

关于守护进程需要强调两点：

其一：守护进程会在主进程代码执行结束后就终止

其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

如果我们有两个任务需要并发执行，那么开一个主进程和一个子进程分别去执行就ok了，如果子进程的任务在主进程任务结束后就没有存在的必要了，那么该子进程应该在开启前就被设置成守护进程。主进程代码运行结束，守护进程随即终止。

import multiprocessing
import time

# 开启进程的 第一种方式
def hi(name, x):
    print("start time", time.ctime())
    print("hi %s" % name)
    time.sleep(x)
    print(" %s end time " % name, time.ctime())


if __name__ == '__main__':
    p = multiprocessing.Process(target=hi, args=("alex", 2))
    p1 = multiprocessing.Process(target=hi, args=("egon", 3))
    p.daemon = True

    p.start()
    p1.start()
    time.sleep(1)
    print("zhu died !", )  # 进程不会等他开的子进程结束才结束
    time.sleep(0.3)
    print(p.is_alive())

互斥锁

进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,而共享带来的是竞争，竞争带来的结果就是错乱

并发运行,效率高,但竞争同一打印终端,带来了打印错乱

from multiprocessing import Process
import os,time
def work():
    print('%s is running' %os.getpid())
    time.sleep(2)
    print('%s is done' %os.getpid())

if __name__ == '__main__':
    for i in range(3):
        p=Process(target=work)
        p.start()

如何控制，就是加锁处理。而互斥锁的意思就是互相排斥，如果把多个进程比喻为多个人，互斥锁的工作原理就是多个人都要去争抢同一个资源：卫生间，一个人抢到卫生间后上一把锁，其他人都要等着，等到这个完成任务后释放锁，其他人才有可能有一个抢到......所以互斥锁的原理，就是把并发改成穿行，降低了效率，但保证了数据安全不错乱。

#由并发变成了串行,牺牲了运行效率,但避免了竞争
from multiprocessing import Process,Lock
import os,time
def work(lock):
    lock.acquire() #加锁
    print('%s is running' %os.getpid())
    time.sleep(2)
    print('%s is done' %os.getpid())
    lock.release() #释放锁
if __name__ == '__main__':
    lock=Lock()
    for i in range(3):
        p=Process(target=work,args=(lock,))
        p.start()

模拟抢票

from multiprocessing import Process, Lock --->  # 进程锁
import time
import json


def search(name):
    count = json.load(open(r"E:第四模块并发编程db.json", "r"))["count"]
    print("<%s>查看了剩余票数为 %s" % (name, count))


def get(name):
    time.sleep(1)
    count_dict = json.load(open(r"E:学并发编程db.json", "r"))
    if count_dict["count"] > 0:
        count_dict["count"] -= 1
        time.sleep(3)
        print("<%s> 购票成功" % name)
        json.dump(count_dict, open(r"E:并发编程db.json", "w"))
    else:
        print("%s fail !" %name)


def task(name,meux):
    search(name)
    with meux:   # 运用了上下文管理器的方法
        get(name)


if __name__ == '__main__':
    # f = open(r"E:并发编程db.json")
    meux = Lock()
    for i in range(10):
        p = Process(target=task, args=("路人<%s>" % i, meux))
        p.start()

进程队列

进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的。

创建队列的类（底层就是以管道和锁定的方式实现）：

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

参数介绍：

maxsize是队列中允许最大项数，省略则无大小限制。
但需要明确：

    1、队列内存放的是消息而非大数据
    2、队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小

主要方法介绍：

q.put方法用以插入数据到队列中。
q.get方法可以从队列读取并且删除一个元素。  

队列的使用

from multiprocessing import Queue

q = Queue(maxsize=3)

q.put("hello word")
q.put(123)
q.put([1, 234])

print("判断队列是否满了", q.full())
q.put()  如果已经满了 再放阻塞住了

print(q.get())
print(q.get())
print(q.get())  # 如果没有数据q.get会卡住 
print("判断队里是否空了", q.empty())

生产者消费者模型介绍

为什么要使用生产者消费者模型

生产者指的是生产数据的任务，消费者指的是处理数据的任务，在并发编程中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者和消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的。

生产者消费者模型实现

from multiprocessing import Process, Queue

import time


def producer(q):
    for i in range(20):
        time.sleep(0.6)
        print("生产的第%s个包子" % (i + 1))
        res = i + 1
        q.put(res)
    # q.put(None)   # 不能这样 因为每一个生产者的生产效率不知道，导致有个生产者生产完了放一个None
    # 这样 消费者就不吃了，但是其实另外一个生产者还有包子再生产,，队列里面有包子其实，但是消费者走了一个


def consumer(q):

    while True:  # while q.size()>0 不能这样 因为刚开始就是0个这样进不了循环导致消费者直接没运行
        res = q.get()
        if res is None:  # 判断如果是none则表明生产者们都生产完了
            break
        time.sleep(1)
        print("消费者吃的第%s包子" % res)


if __name__ == '__main__':
    q = Queue()

    p1 = Process(target=producer, args=(q,))  # 生产者们
    p2 = Process(target=producer, args=(q,))  # 生产者们
    p3 = Process(target=producer, args=(q,))  # 生产者们  3个
    c1 = Process(target=consumer, args=(q,))  # 消费者们  2个
    c2 = Process(target=consumer, args=(q,))  # 消费者们
    p1.start()  # 生产者们启动
    p2.start()
    p3.start()
    c1.start()  # 消费者们启动
    c2.start()
    p1.join()  # 保证等待消费者们执行完毕主程序才往下走
    p2.join()
    p3.join()
    q.put(None)  # 生产完毕之后往队列里面放2 个None  因为 有两个消费者 要发两个终止信息
    q.put(None)
    print('主程序')

JoinableQueue([maxsize]) 的使用

这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

参数介绍

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

方法介绍

JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：

q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常

q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

from multiprocessing import Process, JoinableQueue

import time


def producer(q):
    for i in range(3):
        time.sleep(0.6)
        print("生产的第%s个包子" % (i + 1))
        res = i + 1
        q.put(res)
    q.join()  # 接收信号


def consumer(q):
    while True:  # while q.size()>0 不能这样 因为刚开始就是0个这样进不了循环导致消费者直接没运行
        res = q.get()
        time.sleep(1)
        print("消费者吃的第%s包子" % res)
        q.task_done()  # 这个信号就是说明q这个队列已经全部取完了 结束掉q，
        # 给q.join 发一个信号,如果不发送这个信号，因为生产者不是守护进程，所以主进程就会一直卡住，等着生产者结束


if __name__ == '__main__':
    q = JoinableQueue()
    p1 = Process(target=producer, args=(q,))  # 生产者们
    p2 = Process(target=producer, args=(q,))  # 生产者们
    p3 = Process(target=producer, args=(q,))  # 生产者们  3个
    c1 = Process(target=consumer, args=(q,))  # 消费者们  2个
    c2 = Process(target=consumer, args=(q,))  # 消费者们
    c1.daemon = True  # 这样设置的目的是主进程结束了 消费者没必要存活了
    c2.daemon = True
    p1.start()  # 生产者们启动
    p2.start()
    p3.start()
    c1.start()  # 消费者们启动
    c2.start()
    p1.join()  # 这个是等待生产者都结束，这样主进程才能结束，因为如果不等待生产者结束，
    # 那么主进程一结束,消费者就死了，没有消费者了就不能取包子，于发信号了
    p2.join()
    p3.join()
    print('主程序')