0x01：创建进程

multiprocessing模块就是跨平台版本的多进程模块，提供了一个Process类来代表一个进程对象，这个对象可以理解为是一个独立的进程，可以执行另外的事情。

示例：创建一个进程，执行两个死循环。

from multiprocessing import Process
import time
def run_proc():
    """子进程要执行的代码"""
    while True:
        print("----2----")
        time.sleep(1)
if __name__=='__main__':
    p = Process(target=run_proc)
    p.start()
    while True:
        print("----1----")
        time.sleep(1)

说明

创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动

0x02：方法说明

Process( target [, name [, args [, kwargs]]])

target：如果传递了函数的引用，可以任务这个子进程就执行这里的代码
args：给target指定的函数传递的参数，以元组的方式传递
kwargs：给target指定的函数传递命名参数
name：给进程设定一个名字，可以不设定

Process创建的实例对象的常用方法：

start()：启动子进程实例（创建子进程）
is_alive()：判断进程子进程是否还在活着
join([timeout])：是否等待子进程执行结束，或等待多少秒
terminate()：不管任务是否完成，立即终止子进程

Process创建的实例对象的常用属性：

name：当前进程的别名，默认为Process-N，N为从1开始递增的整数
pid：当前进程的pid（进程号）

实例：

from multiprocessing import Process
import os
from time import sleep
def run_proc(name, age, **kwargs):
    for i in range(10):
        print('子进程运行中，name= %s,age=%d ,pid=%d...' % (name, age, os.getpid()))
        print(kwargs)
        sleep(0.2)
if __name__=='__main__':
    p = Process(target=run_proc, args=('test',18), kwargs={"m":20})
    p.start()
    sleep(1)  # 1秒中之后，立即结束子进程
    p.terminate()
    p.join()

0x03：Pool

开启过多的进程并不能提高你的效率，反而会降低你的效率，假设有500个任务，同时开启500个进程，这500个进程除了不能一起执行之外（cpu没有那么多核），操作系统调度这500个进程，让他们平均在4个或8个cpu上执行，这会占用很大的空间。

如果要启动大量的子进程，可以用进程池的方式批量创建子进程：

def task(n):
    print('{}----->start'.format(n))
    time.sleep(1)
    print('{}------>end'.format(n))
if __name__ == '__main__':
    p = Pool(8)  # 创建进程池，并指定线程池的个数，默认是CPU的核数
    for i in range(1, 11):
        # p.apply(task, args=(i,)) # 同步执行任务，一个一个的执行任务，没有并发效果
        p.apply_async(task, args=(i,)) # 异步执行任务，可以达到并发效果
    p.close()
    p.join()

进程池获取任务的执行结果:
def task(n):
    print('{}----->start'.format(n))
    time.sleep(1)
    print('{}------>end'.format(n))
    return n ** 2
if __name__ == '__main__':
    p = Pool(4)
    for i in range(1, 11):
        res = p.apply_async(task, args=(i,))  # res 是任务的执行结果
        print(res.get())  # 直接获取结果的弊端是，多任务又变成同步的了
       p.close()
    # p.join()  不需要再join了，因为 res.get()本身就是一个阻塞方法

异步获取线程的执行结果:
import time
from multiprocessing.pool import Pool
def task(n):
    print('{}----->start'.format(n))
    time.sleep(1)
    print('{}------>end'.format(n))
    return n ** 2
if __name__ == '__main__':
    p = Pool(4)
    res_list = []
    for i in range(1, 11):
        res = p.apply_async(task, args=(i,))
        res_list.append(res)  # 使用列表来保存进程执行结果
    for re in res_list: 
        print(re.get())
    p.close()

multiprocessing.Pool常用函数解析：

apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
close()：关闭Pool，使其不再接受新的任务；
terminate()：不管任务是否完成，立即终止；
join()：主进程阻塞，等待子进程的退出，必须在close或terminate之后使用；

0x04：进程间不能共享全局变量

from multiprocessing import Process
import os
nums = [11, 22]
def work1():
    """子进程要执行的代码"""
    print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))
    for i in range(3):
        nums.append(i)
        print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))
def work2():
    """子进程要执行的代码"""
    nums.pop()
    print("in process2 pid=%d ,nums=%s" % (os.getpid(), nums))
if __name__ == '__main__':
    p1 = Process(target=work1)
    p1.start()
    p1.join()
    p2 = Process(target=work2)
    p2.start()
    print('in process0 pid={} ,nums={}'.format(os.getpid(),nums))

in process1 pid=2707 ,nums=[11, 22]
in process1 pid=2707 ,nums=[11, 22, 0]
in process1 pid=2707 ,nums=[11, 22, 0, 1]
in process1 pid=2707 ,nums=[11, 22, 0, 1, 2]
in process0 pid=2706 ,nums=[11, 22]
in process2 pid=2708 ,nums=[11]

0x05：进程间通信-Queue

from multiprocessing import Queue
q=Queue(3) #初始化一个Queue对象，最多可接收三条put消息
q.put("消息1") 
q.put("消息2")
print(q.full())  #False
q.put("消息3")
print(q.full()) #True
#因为消息列队已满下面的try都会抛出异常，第一个try会等待2秒后再抛出异常，第二个Try会立刻抛出异常
try:
    q.put("消息4",True,2)
except:
    print("消息列队已满，现有消息数量:%s"%q.qsize())
try:
    q.put_nowait("消息4")
except:
    print("消息列队已满，现有消息数量:%s"%q.qsize())
#推荐的方式，先判断消息列队是否已满，再写入
if not q.full():
    q.put_nowait("消息4")
#读取消息时，先判断消息列队是否为空，再读取
if not q.empty():
    for i in range(q.qsize()):
        print(q.get_nowait())

说明

初始化Queue()对象时（例如：q=Queue()），若括号中没有指定最大可接收的消息数量，或数量为负值，那么就代表可接受的消息数量没有上限（直到内存的尽头）；

Queue.qsize()：返回当前队列包含的消息数量；
Queue.empty()：如果队列为空，返回True，反之False ；
Queue.full()：如果队列满了，返回True,反之False；
Queue.get([block[, timeout]])：获取队列中的一条消息，然后将其从列队中移除，block默认值为True；

1）如果block使用默认值，且没有设置timeout（单位秒），消息列队如果为空，此时程序将被阻塞（停在读取状态），直到从消息列队读到消息为止，如果设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出”Queue.Empty”异常；.

2）如果block值为False，消息列队如果为空，则会立刻抛出”Queue.Empty”异常；

Queue.get_nowait()：相当Queue.get(False)；
Queue.put(item,[block[, timeout]])：将item消息写入队列，block默认值为True；

1）如果block使用默认值，且没有设置timeout（单位秒），消息列队如果已经没有空间可写入，此时程序将被阻塞（停在写入状态），直到从消息列队腾出空间为止，如果设置了timeout，则会等待timeout秒，若还没空间，则抛出”Queue.Full”异常；

2）如果block值为False，消息列队如果没有空间可写入，则会立刻抛出”Queue.Full”异常；

Queue.put_nowait(item)：相当Queue.put(item, False)；
0x06：使用Queue实现进程共享
我们以Queue为例，在父进程中创建两个子进程，一个往Queue里写数据，一个从Queue里读数据： ``` from multiprocessing import Process, Queue import os, time, random

写数据进程执行的代码:

def write(q): for value in [‘A’, ‘B’, ‘C’]: print(‘Put %s to queue…’ % value) q.put(value) time.sleep(random.random())

读数据进程执行的代码:

def read(q): while True: if not q.empty(): value = q.get(True) print(‘Get %s from queue.’ % value) time.sleep(random.random()) else: break

if name==’main‘:

# 父进程创建Queue，并传给各个子进程：
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
# 启动子进程pw，写入:
pw.start()    
# 等待pw结束:
pw.join()
# 启动子进程pr，读取:
pr.start()
pr.join()
print('所有数据都写入并且读完')

```

Python编程学习

6. 进程