python之进程

1 概念

进程：程序执行一次的过程。是程序被读取到内存之中，被操作系统调用时开始生命周期，执行结束即结束生命周期，是一个过程。进程是战占有cpu和内存的。

在linux系统下，创建进程会自动在系统下生成一个PCB（进程控制块）。

PCB：内存中的一小块空间，用来记录进程的各种信息，包括pid，name，调度信息，优先级，状态，虚拟地址等。

pid：操作系统中每一个进程都有唯一的id号，叫做该进程的pid，pid由系统分配。

虚拟地址：是计算机系统内存管理的一种技术，是为了每个进程有足够的内存地址可用。每个进程都拥有独立的4g的虚拟地址空间。

父子进程概念：在系统中的每一个进程（除了初始进程）都是由父进程创建的，每个进程有唯一的父进程，可能有多个子进程，子进程继承了父进程的大部分属性。

2 进程的状态

三态：

　　就绪态：进程具备运行条件，等待系统分配处理器以便运行。

　　运行态：进程占有cpu运行。

　　等待态：又称为阻塞态，或者睡眠态，指进程不具备运行条件，正在等待某个事件的完成。

五态：

　　在三态的基础上，增加了两态

　　新建：创建一个新的进程，直接表现为执行某程序，或者程序中进行子进程的创建。

　　终止：程序执行结束，完成善后。

3 僵尸进程和孤儿进程

　　僵尸进程：进程已经结束，但是系统中依然保存该进程的pcb信息，会占用一定的内存。

　　僵尸进程产生原因：子进程先于父进程退出，父进程没进行处理，这时子进程就会成为僵尸进程。

　　孤儿进程：父进程先退出，此时子进程就会成为孤儿进程，孤儿进程会被系统专有进程进行收养，所以孤儿进程并没有影响。

　　僵尸进程的处理方法：

　　　　1 父进程使用os.wait或者os.waitpid进行处理

　　　　2 创建2级子进程，让一级子进程退出，则2级子进程称为孤儿进程。

　　　　3 父进程使用信号处理方式处理子进程

　　os.wait():

　　功能：阻塞等待子进程的退出，只要该进程有任意子进程退出，则阻塞结束。

　　参数：无 

　　返回值：包含两个元素的元组，第一个是退出的子进程的PID，第二个是退出的子进程相关的退出码

　　os.waitpid(pid, options):　

　　功能：等待子进程退出

　　参数： pid   -1 表示任何一个子进程退出都可以

             　　>0 表示等待指定的子进程退出

                    options  0  表示始终阻塞等待

                   WNOHANG ： 非阻塞等待

　　返回值：和wait相同 包含两个元素的元组，第一个是退出的子进程的PID（如果WNOHANG 则可能是0），第二个是退出的子进程相关的退出码

　　wait() ====  waitpid(-1,0)

4 创建进程 fork

　　1 使用fork（）创建进程

　　fork(）是属于os模块的

　　功能：为当前进程创建一个子进程。

　　参数：无

　　返回值：

　　　　　　1 如果是负数，表示创建进程失败

　　　　　　2 如果是0，表示这是在子进程中返回的

　　　　　　3 如果大于0，表示实在父进程中返回的

　　父进程和子进程都是独立存在的，在执行上相互不受影响。

　　利用父子进程中fork的返回值不同加以区分执行内容是固定的方法。

　　父进程中返回值是子进程的pid号

　　子进程拥有父进程所有的资源，包括fork前已有的变量

　　os.getpid():获取当前进程的pid号

　　os.getppid()：获取当前进程的父进程的pid号

　　os._exit([status])：直接退出当前进程

　　sys.exit([status])：抛出syetemexit异常，如果异常不被处理则进程结束。

　　参数：如果不传或者传入0表示进程正常退出，如果传入一个正整数表示非正常退出，如果传入一个非数字则会打印。

示例:

import os
from time import sleep

pid = os.fork() # 创建子进程
if pid < 0:
    print('create process failed')
elif pid == 0: # 子进程要干的事
    print('this is a child process', os.getpgid())
    print('get my parent id', os.getppid())
    sleep(3)
    print('get my parent id', os.getppid())
else:
    print('this is a parent process', os.getpid())
    print('pid=', pid)
print('------end------')

注意：上段代码只能在linux系统中运行。windows下会报错：

因为fork这个系统命令只有linux才有

5 创建进程 multiprocessing

　　创建步骤:

　　1 确定执行事件，将事件封装成函数

　　2 使用process创建新的进程，将要执行的函数传入到相应的进程对象

　　process参数:

　　　　name:新进程名字

　　　　target；传入的目标函数

　　　　args:以元祖的方式向目标函数进行位置传参

　　　　kwargs:以字典的方式向目标函数进行传参

　　3 使用相应的对象调用start（）启动子进程

　　4 使用相应的对象调用join()函数等待子进程的退出

代码：

import multiprocessing as mp
import os
from time import sleep
from random import randint

def worker():
    sleep(randint(0, 4))
    print(os.getppid(),'-------', os.getpid())
    print('worker')

if __name__ == '__main__':
    jobs = []
    print('this is parent', os.getpid())
    for i in range(5):
        p = mp.Process(target=worker)
        jobs.append(p)
        p.start()
    
    for i in jobs:
        i.join()

一定要加 if __name__ == '__main__':

进程对象p的属性：

p.pid 响应子进程的pid号

p.name 子进程的名字

p.is_alive() 子进程状态

p.start() 启动子进程

p.join([timeout]) 回收子进程的函数，time是一个可选参数，表示超时等待时间

p.demon 默认为False 表示主进程执行结束后不会立即退出，而是等待子进程执行结束后再退出。设置为 True ，那么主进程执行结束后会终止该子进程的执行，并且立即退出。必须在start前设置

import multiprocessing as mp
import os
from time import sleep

a = 100
def worker(sec, message):
    for i in range(3):
        sleep(sec)
        print('the create message', message)
        print('in child process', os.getpid())
    print(a)

if __name__ == '__main__':
    p = mp.Process(target=worker, name='katy', args=(2,), kwargs={'message':'fight world'})
    p.start()
    print('pid', p.pid)
    print('name', p.name)
    print('is_alive', p.is_alive())
    a = 1000
    p.join()
    print('----main process over-----')
    print(a)

上述程序创建了一个带参数的process，也就是给子进程传参

import multiprocessing as mp
from time import sleep

def fun():
    print('child start!')
    sleep(3)
    print('fun is over')

if __name__ == '__main__':
    p = mp.Process(target=fun)
    p.daemon = True
    p.start()
    sleep(4)
    print('----main process over----')

上述程序描述了p.daemon设置为true时的情情况

run()

守护进程：系统中的后台服务进程　　

特点：独立于终端并周期性的执行某周任务，生命周期长，一般随系统启动，随系统结束。

import multiprocessing as mp
import time

class ClockProcess(mp.Process):
    def __init__(self, value):
        mp.Process.__init__(self)
        self.value = value

    def run(self):
        n = 10
        while n > 0:
            print('the time is {}', format(time.ctime()))
            time.sleep(self.value)
            n -= 1

if __name__ == '__main__':
    p = ClockProcess(3)
    p.start()
　　# start 启动时自动调用run方法

上述代码使用了run()方法

6 创建进程 进程池pool

进程池:多个进程执行任务，任务非常多，且执行时间短，需要频繁的创建删除进程

1. 使用 Pool 创建进程池，得到进程池对象
2. 使用 apply_async 将事件放入进程池等待执行
3. 如果池内有空闲进程则会执行等待的事件
4. 使用close关闭进程池，不能够在投放进程
5. 使用join 阻塞等待进程池内现有所有事件都被执行结束后回收子进程（进程池内所有进程均为子进程）

进程池对象（pool）的方法：

异步加载事件到进程池
pool.apply_async(fun,[args = (),[kwargs = {}]])

同步加载事件到进程池
pool.apply(fun,[args = (),[kwargs = {}]])

close() 关闭进程池

join() 阻塞，等待进程池子进程退出。必须在close
后

import multiprocessing as mp
from time import sleep
import os

def worker(msg):
    print(os.getpid(), os.getppid())
    sleep(2)
    print(msg)
    return

if __name__ == '__main__':
    pool = mp.Pool(processes=4)
    result = []
    for i in range(10):
        msg = 'hello %d'% i
        # 向进程池中加载对象
        r = pool.apply(worker, (msg,))
        # r = pool.apply_async(worker, (msg,)) # 异步
        result.append(r)
    # 得到每个事件的返回值 apply方法没有此功能
    # for res in result:
    #     print(res.get())
    # 关闭进程池
    pool.close()
    # 进程池回收
    pool.close()

上述代码分两种情况，注释掉的为异步时候的情况。

pool.map(fun, iterable)
功能上类似内建函数 map 将第二个参数中的每一个数带入到第一函数中进行传参。然后将该事件放入进程池

import time
from multiprocessing import Pool

def run(fn):
    time.sleep(1)
    return fn * fn

test = [1, 2, 3, 4, 5, 6]
print('顺序执行')
s = time.time()
for fn in test:
    run(fn)
e = time.time()
print('执行时间', float(e-s))
print('多任务执行')

if __name__ == '__main__':
    pool = Pool(3)
    #兼顾了apply_async
    r = pool.map(run, test)
    pool.close()
    pool.join()
    e1 = time.time()
    print('执行时间', float(e1-e))

7 进程间的通信 不同的进程间进行消息的传递

1 管道

通过 Pipe 创建，返回值有两个分别表示管道的两端。
当创建时参数为True（默认）时两个返回值均可以进行send和recv操作，当为false时，第一个可进行recv操作第二个可send。
recv函数为阻塞函数，当管道内为空时会阻塞

from multiprocessing import Process,Pipe
import time, os

def fun(child_conn, name):
    child_conn.send("hello" + str(name))
    print(os.getppid(),'------',os.getpid())

if __name__ == '__main__':
    child_conn, parent_conn = Pipe()
    jobs = []

    for i in range(5):
        p = Process(target=fun, args=(child_conn, i,))
        jobs.append(p)
        p.start()

    for k in range(5):
        print(parent_conn.recv())

    for j in jobs:
        j.join()

2 消息队列

　　1.按照先进先出的原则来存取消息，先放入的消息先被取出

　　2. 队列中没有消息则为空队列，这是无法执行取消息的操作。队列中消息个数达到上限则为满队列此时无法存入消息

　　3. 不同进程间，通过向队列存入和获取消息来达到通信的目的

from  multiprocessing import Queue

# 创建消息队列对象
q = Queue(3)
q.put(1)
q.put('hello')
q.put([1, 2, 3, 4])
print(q.full())
# q.put('full',True,timeout = 3)
print(q.qsize())

print(q.get())
print(q.get())
print(q.get())
print(q.qsize())
print(q.empty())
# print(q.get(True, 3))

q = Queue(maxsize)
功能 ：创建消息队列对象
maxsize : 设置消息队列大小，表示最多存多少个消息

q.put(obj,block = True,timeout = None)
功能：向队列中存入消息
obj ：要存入的对象
block：是否是阻塞模式，默认是True 如果设置为False为非阻塞

timeout: 当block = True时为阻塞时间

q.get（）
功能：从队列中取出一个消息
block：默认为True 表示阻塞，设置为false表示为非阻塞则如果队列为空立即返回empty异常
timeout： block= True 时 表示阻塞等待时间，超时则返回异常

q.full()
判断队列是否为满，如果满则返回True否则返回false

q.empty()
判断队列是否为空，如果空则返回True否则返回false

q.qsize()
查看当前队列中消息的个数

from multiprocessing import Process,Queue
import time

def fun(name, q):
    time.sleep(1)
    #每个进程放一个消息
    q.put('hello' + str(name))

if __name__ == '__main__':
    process_list = []
    # 创建队列
    q = Queue()
    for i in range(10):
        p = Process(target = fun,args = (i, q,))
        p.start()
        process_list.append(p)
    for j in process_list:
        j.join()

    #只要队里不空取出消息
    while not q.empty():
        print(q.get())

3 共享内存

特点 ：

1. 效率高的进程间通信方式
2. 安全性上有风险，因为内容的存放是会覆盖原有内容的，所以在使用时很可能已经被篡改
3. 基于2的原因，在使用时经常需要考虑加锁问题

from multiprocessing import Value，Array

1.两种方法使用上基本相同，value在共享内存存放一个数值，array可以存放多个数值，但是类型必须相同

2.任意的进程对共享内存中的数据进行修改后，即其他进程也会获得修改后的数据

3. 两个方法第一个参数相同，都是ctypes，详见表。第二个参数 value为一个相应类型的数值，array可以是数值（表示开辟一个包含多少数据的内存空间填充0），也可以是一个迭代对象（会给距迭代内容开辟空间并且将内容进行填充)

from multiprocessing import Process, Array
import time

def fun(shm, n):
    for i in range(n):
        print(shm[i])
    time.sleep(3)
    shm[2] = 1000
    shm[3] = 5000

if __name__ == '__main__':
    num = 10
    shm = Array('i', num)
    p = Process(target=fun, args=(shm, num,))
    p.start()
    for i in shm:
        time.sleep(2)
        print('---------->', i)
    p.join()

from multiprocessing import Process, Value
import time
import random

def deposit(money):
    for i in range(100):
        time.sleep(0.02)
        money.value += random.randint(1,200)

def withdraw(money):
    for i in range(100):
        time.sleep(0.02)
        money.value -= random.randint(1,100)

if __name__ == '__main__':
    # 将2000按c 中的格式进行转换
    # i 表示转换成什么类型，i为整形
    money = Value('i',2000)

    d = Process(target = deposit,args = (money,))
    d.start()
    w = Process(target = withdraw,args = (money,))
    w.start()
    d.join()
    w.join()
    
    print(money.value)

4 信号

是一种异步的进程间通信方式 信号 有其名称，含义，和默认行为

发送信号
import os

os.kill(pid,sig)
功能：向一个进程发送一个信号
pid ： 向哪个进程发送，该进程的PID号
sig ： 发送什么信号 使用 signal.signum

import os
import signal 

os.kill(7319,signal.SIGKILL)

signal.alarm(sec)
功能：向自身发送一个信号
sec： 在sec秒后信号会被发送

import signal
import time

#向自身发送一个时钟信号
signal.alarm(4)
time.sleep(3)
signal.alarm(8) #会重置前一个时钟

while True:
    time.sleep(1)
    print('我还在蹦跶......')

signal.pause()
功能：挂起等待一个信号

signal.signal(signum,handler)
功能：处理一个信号
参数： signum ： 要处理的信号

import signal

signal.alarm(5)

signal.pause()

print("(ˇˍˇ) 想～执行")

handler： 对信号的处理方法
处理方法：忽略该信号 SIG_IGN
使用默认方法执行 SIG_DFL
使用指定方法执行 function

import signal 

signal.alarm(6)

signal.signal(signal.SIGINT,signal.SIG_IGN)

signal.signal(signal.SIGALRM,signal.SIG_DFL)

signal.pause()

异步处理 ： 在某一时刻使用signal扑捉信号，会告知内核帮进程监控，而不是阻塞等待。在进程的生命周期内，只要有该信号发送进来就会处理。

僵尸进程处理
父进程在子进程执行结束前加入：
signal.signal(signal.SIGCHLD,signal.SIG_IGN)

同步和互斥

临界资源 ： 对多个进程或线程可见，容易产生争夺的资源（如共享内存）称之为临界资源

临界区 ： 对临界资源进行操作的代码段，称之为临界区

同步 ： 同步是一种制约关系，为完成某种任务而建立两个或多个进程，进程间协调而有次序的等待，传递信息，完成工作。这种制约源于进程间的合作

互斥 ：互斥是一种间接的制约，当一个进程进入临界区进行加锁，其他进程此时无法操作临界资源，只有当该进程结束对临界资源的使用后，进行解锁，其他进程才可以使用。这种技术往往是通过阻塞完成的

同步互斥方法

Event

e = Event() 创建事件对象

e.is_set() 判断事件是否被设置

e.set() 对事件对象进行设置

e.wait(2) 阻塞等待时间被设置 （参数表示超时时间）

from multiprocessing import Event

e = Event()

print(e.is_set()) #没有set打印false

e.set() # 对事件对象进行设置

e.wait(2)  # 阻塞等待

print(e.is_set())

print("wait ......")

from multiprocessing import Event,Process
import time

def wait_event(e):
    print('wait for event setting')
    #阻塞等待事件setting
    e.wait()
    print('wait_for_event_1:',e.is_set())

def wait_event_timeout(e):
    print('wait for event setting or timeout')
    #阻塞等待事件setting
    e.wait(2)
    print('wait_for_event_2:',e.is_set())

if __name__ == '__main__':
    #创建事件对象
    e = Event()

    p1 = Process
    (name = 'block',target = wait_event, args=(e,))
    p1.start()
    p2 = Process
    (name='non-block',target=wait_event_timeout, args=(e,))
    p2.start()

    print("main: setting the event")
    time.sleep(3)
    e.set()
    print("event is set")

Lock
from multiprocessing import Lock
lock = Lock()

lock.acquire() 上锁
lock.release() 解锁

with lock： 上锁

import multiprocessing  
import sys,time

def worker1(stream):
    lock.acquire()
    for i in range(5):
        time.sleep(1)
        stream.write('Lock acquired via
')
    lock.release()

def worker2(stream):
    with lock:
        for i in range(5):
            time.sleep(1)
            stream.write('Lock acquired directly
')

#创建锁对象
lock = multiprocessing.Lock()

w1 = multiprocessing.Process
(target = worker1,args = (sys.stdout,))

w2 = multiprocessing.Process
(target = worker2,args = (sys.stdout,))

w1.start()
w2.start()

w1.join()
w2.join()