进程理论，进程创建，进程同步，进程间通信，进程池

一  进程理论

1.操作系统原理

2. 进程基础

2.1 定义

　　进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。在当代面向线程设计的计算机结构中，进程是线程的容器，线程是程序的基本执行实体。程序是指令、数据及其组织形式的描述，进程是程序的实体。

　　第一，进程是一个实体。每一个进程都有它自己的地址空间，一般情况下，包括文本区域（text region）、数据区域（data region）和堆栈（stack region）。文本区域存储处理器执行的代码；数据区域存储变量和进程执行期间使用的动态分配的内存；堆栈区域存储着活动过程调用的指令和本地变量。

　　第二，进程是一个“执行中的程序”。程序是一个没有生命的实体，只有处理器赋予程序生命时（操作系统执行之），它才能成为一个活动的实体，我们称其为进程。进程是操作系统中最基本、重要的概念。是多道程序系统出现后，为了刻画系统内部出现的动态情况，描述系统内部各道程序的活动规律引进的一个概念,所有多道程序设计操作系统都建立在进程的基础上

2.2  进程调度

　　先来先服务（FCFS）调度算法最简单，该算法比较有利于长作业（进程），而不利于短作业（进程），适合于CPU繁忙型作业，而不利于I/O繁忙型的作业（进程）。

　　短作业（进程）优先调度算法（SJ/PF）是指对短作业或短进程优先调度的算法，其对长作业不利；不能保证紧迫性作业（进程）被及时处理；作业的长短只是被估算出。

　　时间片轮转(Round Robin，RR)法的基本思路是让每个进程在就绪队列中的等待时间与享受服务的时间成比例。在时间片轮转法中，需要将CPU的处理时间分成固定大小的时间片，如果一个进程在被调度选中之后用完了系统规定的时间片，但又未完成要求的任务，则它自行释放自己所占有的CPU而排到就绪队列的末尾，等待下一次调度。

　　多级反馈队列调度算法应用最广，过程如下：设置多个就绪队列，并为各个队列赋予不同的优先级。在优先权愈高的队列中，为每个进程所规定的执行时间片就愈小。当一个新进程进入内存后，首先将它放入第一队列的末尾，按FCFS原则排队等待调度。当轮到该进程执行时，如它能在该时间片内完成，便可准备撤离系统；如果它在一个时间片结束时尚未完成，调度程序便将该进程转入第二队列的末尾，再同样地按FCFS原则等待调度执行，以此类推下去。

2.3 并行与并发

• 并行 : 并行是指两者同时执行，多核。
• 并发 : 并发是指资源有限的情况下，两者交替轮流使用资源，提高效率，单核。

2.4  进程三种状态

　　就绪(Ready)状态：当进程已分配到除CPU以外的所有必要的资源，只要获得处理机便可立即执行；

　　执行/运行（Running）状态：当进程已获得处理机，其程序正在处理机上执行；

　　阻塞(Blocked)状态：正在执行的进程，由于等待某个事件发生而无法执行时，便放弃处理机（cpu）而处于阻塞状态。引起进程阻塞的事件可有多种，例如，等待I/O完成（input）、申请缓冲区不能满足、等待信件(信号)等。

 

2.5 同步与异步

　　同步就是一个任务的完成需要依赖另外一个任务时，只有等待被依赖的任务完成后，依赖的任务才能算完成，这是一种可靠的任务序列。要么成功都成功，失败都失败，两个任务的状态可以保持一致。

　　异步是不需要等待被依赖的任务完成，只是通知被依赖的任务要完成什么工作，依赖的任务也立即执行，只要自己完成了整个任务就算完成了。至于被依赖的任务最终是否真正完成，依赖它的任务无法确定，所以它是不可靠的任务序列。

2.6  进程创建与结束

创建进程的方式

• 系统初始化（查看进程linux中用ps命令，windows中用任务管理器，前台进程负责与用户交互，后台运行的进程与用户无关，运行在后台并且只在需要时才唤醒的进程，称为守护进程，如电子邮件、web页面、新闻、打印）
• 一个进程在运行过程中开启了子进程（如nginx开启多进程，os.fork,subprocess.Popen等）
•  用户的交互式请求，而创建一个新进程（如用户双击暴风影音）
• 一个批处理作业的初始化（只在大型机的批处理系统中应用）

 结束进程的方式

• 正常退出（自愿，如用户点击交互式页面的叉号，或程序执行完毕调用发起系统调用正常退出，在linux中用exit，在windows中用ExitProcess）
• 出错退出（自愿，python a.py中a.py不存在）
• 严重错误（非自愿，执行非法指令，如引用不存在的内存，1/0等，可以捕捉异常，try...except...）
• 被其他进程杀死（非自愿，如kill -9）

二  进程创建

　　multiprocessing模块几乎包含了和进程有关的所有子模块。大致分为四个部分：进程创建，进程同步，进程池，进程之间数据共享。

　　创建进程包含multiprocessing.Process模块

1. 模块简介

Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）

参数介绍：必须使用关键字传参
group参数未使用，值始终为None
target表示调用对象，即子进程要执行的任务
args表示调用对象的位置参数元组，args=(1,2,'egon',)
kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
name为子进程的名称

方法介绍：
p.start()：启动进程，并调用该子进程中的p.run() 
p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法  
p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive():如果p仍然运行，返回True
p.join([timeout]):主进程等待p终止。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程  

属性介绍：
p.daemon：默认值为False，如果设为True，代表p为守护进程，当p的父进程终止时，p也随之终止（正常程序是主进程等待子进程结束后再一起结束），p不能创建自己的新进程，必须在p.start()之前设置
p.name:进程的名称
p.pid：进程的pid，相比于os.getpid（）
p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）

注意：在Windows操作系统中由于没有fork(linux操作系统中创建进程的机制)，在创建子进程的时候会自动import启动它的这个文件，而在import的时候又执行了整个文件。
因此如果将process()直接写在文件中就会无限递归创建子进程报错。所以必须把创建子进程的部分使用if __name__ ==‘__main__’ 判断保护起来。

2.  开启进程的两种方法

2.1 函数法（常用）

from multiprocessing import Process
import os
def func():
    print('func',os.getpid())
if __name__ == '__main__':
    a = Process(target=func,)
    a.start()
    print('main',os.getpid())
　　print（a.name）

main 6480；func 4424；Process -1　　　　#name未定义之前，采用父类名称（Process）

2.2 类继承方法

from multiprocessing import Process
import os
def func(n):
    print('func:', os.getpid())
class A(Process):
    def __init__(self, target, args):
        super().__init__(target=target, args=args)  # 源码中默认值都是空
        print('A:', os.getpid())
    def fun2(self):
        print('func2:', os.getpid())
    def run(self):  # run函数时开启新进程，其他内部函数不能开启新进程
        print('run:', os.getpid())
if __name__ == '__main__':
    a = A(target=func, args=(100,))  # 子进程中的初始化函数还是在主进程中完成的
    a.start()  # start调用的是run，
    a.fun2()
    print('main:', os.getpid())
    print('a.pid():', a.pid)
    print(a.name)

A: 7824
func2: 7824
main: 7824
a.pid(): 10112
A-1　　　　#子进程名称，A为父类名称，-1表示第一子进程
run: 10112

3.  join（）

　　主进程等待子进程结束后再继续执行，join只能join住start开启的进程，而不能join住run开启的进程

from multiprocessing import Process
import time
def func(i):
    print('这里是第%s个子进程'%(i))
    time.sleep(1)

if __name__ == '__main__':
    p_l = []
    for i in range(5):
        p = Process(target = func,args=(i,))
        p.start()
        p_l.append(p)
    #     p.join()　　　　　　　　　　#只有此句，串行
     [i.join() for i in p_l]　　　　#只有此句，子程序并行，主程序与子程序串行（常用）
    print('这里是main爸爸')　　　　　#上两句都没有，并行

4.  daemon守护进程

父进程创建守护进程，需注意：

• 守护进程会在父进程代码执行结束后终止，并非整个程序结束，也无需等待子进程结束；此处与线程不一致，线程等待非守护进程的子进程结束后再终止；
• 守护进程内无法再开启子进程，否则抛出异常
• 必须在start（）之前设置

from multiprocessing import Process
import time

def func2():
    print('子进程2开始执行')
    time.sleep(2)
    print('子进程2结束执行')
def func1():
    print('子进程开始执行')
    time.sleep(2)
    print('子进程结束执行')
if __name__ == '__main__':
    print('主进程开始执行')
    p1 = Process(target=func1,)
    p2 = Process(target=func2)
    p1.daemon = True# 将p1 设置为守护进程，此代码一定要在start之前设置。
    p1.start()
    p2.start()
    time.sleep(1)
    print('主进程结束执行')# 当主进程打印完这句话，代表主进程结束，守护进程p1肯定随之结束
    # 但是p2 不是守护进程，不会结束，所以此时程序（也就是主进程）会等待p2结束之后才结束。

主进程开始执行；子进程开始执行；子进程2开始执行；主进程结束执行；子进程2结束执行

5. socket-tcp协议并发实现

　　同时连接多个client，相当于socketserver

from multiprocessing import Process
import socket
from socket import SOL_SOCKET,SO_REUSEADDR
SERVER_ADDR = ('127.0.0.1',8080)
sk = socket.socket()
sk.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
sk.bind(SERVER_ADDR)
sk.listen(5)
def func(conn):
    while 1:
        try:
            msg_r = conn.recv(1024).decode('utf-8')
            print(msg_r)
            if not msg_r:break
            conn.send(msg_r.upper().encode('utf-8'))
        except:
            break
if __name__ == '__main__':
    while 1:
        conn,addr = sk.accept()
        p = Process(target=func,args=(conn,))
        p.start()
    sk.close()

　　客户端

import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8080))
while 1:
    msg_s = input('>>>')
    if not msg_s:continue
    sk.send(msg_s.encode('utf-8'))
    print(sk.recv(1024).decode('utf-8'))
sk.close()

6. 进程间相互独立

　　进程间数据是不共享

def func():
    global n
    n = 0
    print('子进程内 n = %s'%n)
if __name__ == '__main__':
    n = 100
    p = Process(target=func)
    p.start()
    print('主进程内 n = %s'%n)

主进程内 n = 100；子进程内 n = 0　　　　#主进程和子进程的数据不共享

三  进程同步

进程同步包含了multiprocessing.Lock、multiprocessing.Semaphore、multiprocessing.Event等模块

1. Lock互斥锁

应用：当多个进程使用同一份数据资源的时候，就会引发数据安全或顺序混乱问题。

对锁的理解：

• 锁的实质是实现多进程间部分代码串行，不影响未锁定代码
• join（）是实现主进程和子进程的串行

from multiprocessing import Process,Lock
import time
def func(l):
    l.acquire()
    time.sleep(5)　　　　#锁并非只锁定资源，实质是各个进程间锁定代码串行，不影响未锁定代码
    # print('bbb')
    l.release()
if __name__ == '__main__':
    l = Lock()
    l2 =Lock()　　　　#不同的锁之间不执行串行
    a= Process(target=func,args=(l,))
    a.start()
    l.acquire()
    time.sleep(4)
    # print('aaa')
    l.release()

 多人抢票示例：

from multiprocessing import Process, Lock
import json
import time
def check(i, l):
    with open('a.txt', 'r', encoding='utf-8') as f:
        dic = json.load(f)
    print('第%s个人在查票，余票为%s' % (i, dic['c']))
    pay(i, l)
def pay(i, l):
    l.acquire()
    with open('a.txt', 'r', encoding='utf-8') as f:
        dic = json.load(f)
    time.sleep(0.5)  # 模拟网络延迟，当购买过程中也会有网络延迟
    if dic['c']:
        print('第%s个人买到票了 ' % i)
        dic['c'] -= 1
    else:
        print('第%s个人没买到票' % i)
    with open('a.txt', 'w') as f:
        json.dump(dic, f)
    l.release()
if __name__ == '__main__':
    l = Lock()
    for i in range(5):
        p = Process(target=check, args=(i + 1, l))
        p.start()

2. Semaphore信号量

　　Semaphore信号量又可称为多钥匙锁。Lock属于互斥锁，也就是一把钥匙配备一把锁，同时只允许锁住某一个数据。而信号量则是多把钥匙配备多把锁，也就是说同时允许锁住多个数据。信号量同步基于内部计数器，用户初始化一个计数器初值（比如上述例子中就初始化为5），每调用一次acquire()，计数器减1；每调用一次release()，计数器加1。当计数器为0时，acquire()调用被阻塞。

from multiprocessing import Semaphore
from multiprocessing import Process
import time
import random
def sing(i,se):
    se.acquire()# 每次进来一位客人，信号量内部计数器减1
    print('%s进入小黑屋'%i)
    time.sleep(random.randint(1,3))
    print('%s交钱走人'%i)
    se.release()# 每次离开一位客人，信号量内部计数器加1
if __name__ == '__main__':
    se = Semaphore(5)# 初始化5把钥匙配备5把锁
    for i in range(10): # 模拟10个人要进入小黑屋子
        p = Process(target=sing,args=(i,se))
        p.start()

3.  Event 事件

　　主要用于主进程控制其他进程的执行，事件主要提供了三个方法 set、wait、clear。事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

• clear：将“Flag”设置为False，wait阻塞
• set：将“Flag”设置为True，wait不阻塞
• is_set：返回全局‘Flag’的bool值

from multiprocessing import Process, Event
import time
def traffic_signal(e):
    while 1:
        if e.is_set():
            time.sleep(5)
            e.clear()
            print('red light')
        else:
            time.sleep(5)
            e.set()
            print('green light')
def car(e,i):
    print('car%s is running'%i)
if __name__ == '__main__':
    e = Event()
    p = Process(target=traffic_signal,args=(e,))
    p.start()
    for i in range(50):
        e.wait()
        c = Process(target=car,args=(e,i))
        c.start()
        time.sleep(1)

四  进程间通信

1. Queue队列

　　概念：创建共享的进程队列，Queue是多进程安全的队列（底层自动加锁），可以使用Queue实现多进程之间的数据传递。底层队列使用管道和锁定实现。

import queeue，这是线程队列，不能进行进程间的信息交互

• 栈：先进后出，FILO
• 队列：先进先出，FIFO

Queue([maxsize]) 

q = Queue([maxsize])

q.get( [ block [ ,timeout ] ] ) 
返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。
timeout是可选超时时间，用在阻塞模式中。如果在指定的时间间隔内没有项目可用，将引发Queue.Empty异常。

q.get_nowait( ) 
同q.get(False)方法。

q.put(item [, block [,timeout ] ] ) 
将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。
timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。

q.qsize() 
返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。


q.empty() 
如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。

q.full() 
如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。

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

from multiprocessing import Queue,Process
import time
import random
def get_func(q,consumer):
    while 1:
        time.sleep(random.randint(1, 3))
        info = q.get()
        if info == None:break# 当获取到结束生产的标识时，消费者自动break出死循环
        print('33[32m%s 拿走了%s 33[0m'%(consumer,info))
def put_func(q,product):
    for i in range(5):
        info = '%s %s号'%(product,i)
        q.put(info)
        print('33[31m生产了%s 33[0m' % info)
        time.sleep(random.randint(1, 3))
if __name__ == '__main__':
    q = Queue(5)
    con1 = Process(target=get_func, args=(q,'Alex'))
    con2 = Process(target=get_func, args=(q,'WuSir')) 
    pro1 = Process(target=put_func, args=(q,'苍老师版'))
    pro2 = Process(target=put_func, args=(q,'小泽玛雅丽版'))
    pro3 = Process(target=put_func, args=(q,'岛国米饭保你爱版'))
    pro1.start()
    pro2.start()
    pro3.start()
    con1.start()
    con2.start()
    pro1.join()# 让主进程可以获取到生产者结束生产
    pro2.join()# 主进程必须等待所有生产者生产完毕后才可以放结束生产的信号
    pro3.join()
    q.put(None)# 有几个消费者就应该给几个信号，确定结束时间
    q.put(None)

2. JoinableQueue([maxsize]) 

JoinableQueue的实例除了与Queue对象相同的方法之外，还具有以下方法：
q.task_done() 
消费者使用此方法发出信号，表示队列中放入的数据已经被处理。如果调用此方法的次数大于从队列中获取的数据数量，将引发ValueError异常。每次get，都+1，并且返回给q.join。
q.join() 
自动记录加入队列的元素个数，生产者将使用此方法进行阻塞，直到队列中所有项目均被处理。阻塞将持续到消费者为队列中的每个数据均调用q.task_done()方法为止。

　　生产者消费模型，调用JoinableQueue的作用是明确知道结束的时间

from multiprocessing import JoinableQueue,Process
import time, random
def get_func(q,consumer):
    while 1:
        time.sleep(random.randint(1, 2))
        info = q.get()
        print('33[32m%s 拿走了%s 33[0m'%(consumer,info))
        q.task_done()# 消费者每消费一个数据，就要返回一次标识给q.join，证明数据已经被消费
def put_func(q,product):
    for i in range(5):
        info = '%s %s号'%(product,i)
        q.put(info)
        print('33[31m生产了%s 33[0m' % info)
        time.sleep(random.randint(3, 4))
    q.join()# 生产完毕，使用此方法进行阻塞，直到队列中所有数据均被处理。
if __name__ == '__main__':
    q = JoinableQueue(5)
    con1 = Process(target=get_func, args=(q,'Alex'))
    con2 = Process(target=get_func, args=(q,'WuSir'))
    pro1 = Process(target=put_func, args=(q,'苍老师版'))
    pro2 = Process(target=put_func, args=(q,'小泽玛雅丽版'))
    pro3 = Process(target=put_func, args=(q,'岛国米饭保你爱版'))
    con1.daemon = True    　　　　　　#get会阻塞，不能结束
    con2.daemon = True
    l_p = [pro1,pro2,pro3,con1,con2]
    for i in l_p:
        i.start()
    pro1.join()　　　　　#不能省略，语句执行完即守护进程结束，造成异常
    pro3.join()
    pro2.join()
# 此代码的大体逻辑：主进程开启3个生产者和2个消费者，合计6个进程在并发执行。
# 主进程等待 pro1 pro2 pro3的执行完毕
# pro1 pro2 pro3 进程中都有q.join()，会等待消费者con1 con2消费完所以数据
# 即 ：主进程在等pro1 pro2 pro3 ，而pro1 pro2 pro3 在等 con1 con2
# 此代码中消费者进程con1 con2 会一直阻塞在q.get等待接收数据，而造成程序不会结束

3. 其他方法

• Pipe
• Value
• Manager

五  进程池

　　进程池中的最优数：os.cpu_count（）+1；线程数：os.cpu_count（）*5

1. 参数解析

p = Pool([numprocess  [,initializer [, initargs]]]):创建进程池

1 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
2 initializer：是每个工作进程启动时要执行的可调用对象，默认为None
3 initargs：是要传给initializer的参数组

p.apply(func [, args [, kwargs]]):　同步执行，args为iterable，为元组的形式，返回结果为func的返回值

p.apply_async(func [, args [, kwargs]],callback=None):异步执行，args为iterable，元组的形式，返回结果为ApplyResult对象，用get（）获取具体值。只有异步才有回调函数，回调函数是主进程调用的。
'''此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将结果传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。'''

p.map( func, iterable):将iterable序列中每一个元素当成参数传递给func，异步执行func函数。返回值为func的返回值，形式为列表。

p.close():禁止再向进程池中添加新的进程任务。如果所有操作持续挂起，它们将在工作进程终止前完成

P.jion():等待所有进程退出。此方法只能在close（）或teminate()之后调用

2. 注意事项

2.1 多进程、进程池同步、进程池异步效率对比

异步效率会高一些，因为省掉网路延迟等；同步还是一个接一个的执行，异步是多个一起执行。

from multiprocessing import Pool,Process
import os,time
def func(i):
    return i**5
    # print(i**5)
    # time.sleep(1)
if __name__ == '__main__':
    start = time.time()
    pool = Pool(os.cpu_count()+1)
    # pool.map(func,[i for i in range(100)])

    for i in range(100):
        pool.apply(func,(i,))
　　　　# pool.apply_async(func,(i,)).get（）　　　返回结果为ApplyResult对象，用get（）调用，get会阻塞，会使进程变成同步，一个一个添加。

    # p_all=[]
    # for i in range(100):
    #     p = Process(target=func,args=(i,))
    #     p_all.append(p)
    #     p.start()
    # for p in p_all:
    #     p.join()

    pool.close()
    pool.join()
    print(time.time()-start)

#同步：0.835047721862793，不需要用添加close，join，此处差异不明显是因为没有i/o和网络延时的影响
#异步：0.838047981262207，需要添加close，join
#多进程：13.786788702011108，需要添加join
#计算密集时，三者都没有单进程块

2.2 进程池内：同步是普通进程，异步是守护进程

from multiprocessing import Pool, Process
import os, time
def func(i):
    print(os.getpid())
    time.sleep(1)
if __name__ == '__main__':
    pool = Pool(5)
    for i in range(20):
        # pool.apply(func, (i,))  #同步不需要加入close和join，内部为普通进程
        pool.apply_async(func, (i,))　　　#异步必须加入close和join，内部为守护进程
    pool.close()
    pool.join()

5652；1164；6956；5760；2432；5652；1164；6956；5760；2432　　#五个进程号

2.3 回调函数

• 回调函数只有异步中有
• 回调函数是主程序调用的，在线程中是子线程调用的

from multiprocessing import Pool, Process
import os, time
def func(i):
    print('func:',os.getpid())
    return i**2
def call_back(var):
    print(var,os.getpid())

if __name__ == '__main__':
    pool = Pool(5)
    pool.apply_async(func, (10,),callback=call_back)
    pool.close()
    pool.join()
    print('main:',os.getpid())

func: 6060
100 5708
main: 5708