Python3 从零单排24_多进程

　　进程：正在进行的一个过程或者说一个任务。而负责执行任务则是cpu。 程序仅仅只是一堆代码而已，而进程指的是程序的运行过程

　　1.开启进程的两种方式：

from multiprocessing import Process
import time
import os

# 方式一：
def task(name):
    print("%s is running...,pid: %s; ppid: %s " % (name, os.getpid(), os.getppid()))  # 拿到进程pid以及父进程pid
    time.sleep(1)
    print("%s is done..." % name)

# 方式二
class CreateProcess(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name  # 如果没指定名字，会有自己的名称，类名+编号

    def run(self):
        print("%s is running...,pid: %s; ppid: %s " % (self.name, os.getpid(), os.getppid()))
        time.sleep(3)
        print("%s is done..." % self.name)


if __name__ == "__main__":
    # 方式一：
    p1 = Process(target=task, args=("方式一子进程",))
    p1.start()  # 告诉操作系统帮忙开启新的子进程，操作系统开辟一个新的内存空间给子进程使用，并把主进程的内存数据复制到子进程的内存空间里
    print(p1.name)  # 进程名称，Process-编号 也可以自己指定
    p4 = Process(target=task, name="我是方式一的子进程4444", args=("方式一子进程4",))
    print(p4.name)

    # 方式二：
    p2 = CreateProcess("方式二子进程1")
    p3 = CreateProcess("方式二子进程2")
    p2.start()
    p3.start()

    print(p1.is_alive())  # true
    print(p3.is_alive())  # true
    p3.terminate()  # 杀调子进程，将子进程变成僵尸进程。实质是发送杀掉进程命令给操作系统，操作系统回收内存资源，保留子进程状态信息
    p1.join()  # 等待子进程p1运行结算才往下运行
    print("after terminate p3:", p3.is_alive())  # false
    print(p1.pid, p2.pid, p3.pid)  # 查看子进程的pid
    print(p1.is_alive())  # 判断进程存活状态，因为上面join了，所以p1已经是结束了的，只是保留了状态，所以false
    print("主进程结束！ pid: %s" % os.getpid())  # 拿到进程pid

　　僵尸进程：子进程运行完后，内存空间被操作系统收回，但是进程状态需要保留给父进程查看； 也就是说子进程实际上是死了，但是还留有状态信息给父进程可查，等父进程也运行完后，才会清理掉这个已经死掉的子进程状态信息。

　　孤儿进程：父进程挂掉后，子进程还在运行，这个时候就由操作系统的 init(linux操作系统中国呢所有进程的父进程) 进程监管/回收这些子进程

　　2.守护进程 ：子进程设置为主进程的守护进程，主进程运行完(重点，主进程运行完！！！)，子进程也跟着死掉。

from multiprocessing import Process
import time
class my_process_thread(Process):
    def run(self):
        print("%s is running" % self.name)
        time.sleep(1)
        print("%s is done" % self.name)


if __name__ == "__main__":
    p1 = my_process_thread()
    p1.daemon = True  # 设置为守护线程后，这里主线程运行完，p1就死掉了，不会打印进程1
    p1.start()
    p2 = my_process_thread()
    p2.start()
    print("主线程 is over !")

　　3.互斥锁

# 问题：竞争资源会出现数据错乱，比如多个进程竞争终端，都要打印数据，大家都在打印，数据就乱了。
from multiprocessing import Process
import time
class my_process_thread(Process):
    def run(self):
        print("%s is running 第一行" % self.name)
        time.sleep(1)
        print("%s is running 第二行" % self.name)
        time.sleep(1)
        print("%s is running 第三行" % self.name)
        time.sleep(1)
        print("%s is done" % self.name)


if __name__ == "__main__":
    for i in range(3):
        p = my_process_thread()
        p.start()
    print("主线程 is over !")



# 主线程 is over !
# my_process&thread-1 is running 第一行
# my_process&thread-2 is running 第一行
# my_process&thread-3 is running 第一行
# my_process&thread-1 is running 第二行
# my_process&thread-2 is running 第二行
# my_process&thread-3 is running 第二行
# my_process&thread-1 is running 第三行
# my_process&thread-3 is running 第三行Myprocess-2 is running 第三行
# 
# my_process&thread-1 is done
# my_process&thread-3 is done
# my_process&thread-2 is done
'''

# 互斥锁的意义在于将并发变回串行，一个进程用完下个进程才可以用,数据就有序了，但是效率降低，取决你需要数据安全性还是效率。
'''
from multiprocessing import Process, Lock
import time
class my_process&thread(Process):
    def __init__(self, lock):
        super().__init__()
        self.lock = lock

    def run(self):
        self.lock.acquire()  # 申请锁
        print("%s is running 第一行" % self.name)
        time.sleep(1)
        print("%s is running 第二行" % self.name)
        time.sleep(1)
        print("%s is running 第三行" % self.name)
        time.sleep(1)
        print("%s is done" % self.name)
        self.lock.release()  # 释放锁


if __name__ == "__main__":
    lock = Lock()   # 实例化锁，子进程拿到锁才可以运行
    for i in range(3):
        p = my_process&thread(lock)
        p.start()
    print("主线程 is over !")


# 主线程 is over !
# my_process&thread-1 is running 第一行
# my_process&thread-1 is running 第二行
# my_process&thread-1 is running 第三行
# my_process&thread-1 is done
# my_process&thread-2 is running 第一行
# my_process&thread-2 is running 第二行
# my_process&thread-2 is running 第三行
# my_process&thread-2 is done
# my_process&thread-3 is running 第一行
# my_process&thread-3 is running 第二行
# my_process&thread-3 is running 第三行
# my_process&thread-3 is done

　　4.join和互斥锁的区别
　　join只能让整个子进程串行，互斥锁可以让局部代码串行(比如修改数据部分串行，其他查询可以继续并行)

from multiprocessing import Process,Lock
import json
import time

def search(name):
    time.sleep(1)
    dic=json.load(open('db.txt','r',encoding='utf-8'))
    print('<%s> 查看到剩余票数【%s】' %(name,dic['count']))


def get(name):
    time.sleep(1)
    dic=json.load(open('db.txt','r',encoding='utf-8'))
    if dic['count'] > 0:
        dic['count']-=1
        time.sleep(3)
        json.dump(dic,open('db.txt','w',encoding='utf-8'))
        print('<%s> 购票成功' %name)
    else:
        print('<%s> 购票失败' %name)

def task(name,):
    search(name)
    # mutex.acquire()
    get(name)
    # mutex.release()

if __name__ == '__main__':
    # mutex=Lock()
    for i in range(10):
        p=Process(target=task,args=('路人%s' %i,))
        p.start()
        p.join()

　　5.队列

　　进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的
　　创建队列的类（底层就是以管道和锁定的方式实现）：
　　Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。

　　maxsize是队列中允许最大项数，省略则无大小限制。但需要明确：
　　　　1、队列内存放的是消息而非大数据
　　　　2、队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小

from multiprocessing import Queue
q = Queue(3)

q.put(123)  # 往队列里塞一个数据，可以是任何数据类型
q.put([4, 5, 6])
print(q.full())  # 判断队列大小是否已经达到最大项数，这里才塞了俩数据，所以false
q.put({"a": "apple"})
print(q.full())  # 已经达到最大项数了，True
print(q.empty())  # 判断队列里的是否不存在数据了，这里队列了有三数据，所以为false

print(q.get())  # 根据先进先出原则，从队列里取一个数据
print(q.get())
print(q.get())
print(q.empty())  # True


# 生产者/消费者模型
from multiprocessing import Queue
from multiprocessing import Process
import time


def producer(name, q):
    for i in range(5):
        time.sleep(0.5)
        res = "%s生产的馒头%s" % (name, i)
        q.put(res)
        print(res)


def consumer(name, q):
    while True:
        time.sleep(0.2)
        res = q.get()
        if not res:
            break
        print("%s 吃了 %s" % (name, res))


if __name__ == "__main__":
    q = Queue()
    p_lis = []
    for i in range(3):
        p = Process(target=producer, args=("xg%s" % i, q))
        p_lis.append(p)
        p.start()
    s = Process(target=consumer, args=("yy", q))
    s.start()
    for p in p_lis:
        p.join()
    q.put(None)  # 注意，有几个消费者就要发送几个None
    print("over")

　　6.JoinableQueue 的使用 可以用join方法的Queue

　　JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：
　　q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
　　q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

from multiprocessing import JoinableQueue
from multiprocessing import Process
import time


def producer(name, q):
    for i in range(5):
        time.sleep(0.5)
        res = "%s生产的馒头%s" % (name, i)
        q.put(res)
        print(res)
    q.join()  # 等待q结束(即q队列数据全部调用了q.task_done()方法，才往下继续运行代码)


def consumer(name, q):
    while True:
        time.sleep(0.2)
        res = q.get()
        print("%s 吃了 %s" % (name, res))
        q.task_done()  # 每消费一个数据，调用一次task_done方法，为了给join方法用


if __name__ == "__main__":
    q = JoinableQueue()
    p_lis = []
    for i in range(3):
        p = Process(target=producer, args=("xg%s" % i, q))
        p_lis.append(p)
        p.start()
    s = Process(target=consumer, args=("yy", q))
    s.daemon = True  # 主线程运行完，就销毁掉生产者的进程
    s.start()
    for p in p_lis:
        p.join()
    print("over")  # 走到这一步说明，生产者已经运行完了，生产者运行完，代表队列里的数据都调用了task_done方法，也就是消费者运行也结束了