Python进程、进程池以及进程间通信

怼烎@ 2022-06-03 00:11 337阅读 0赞

## 概念 ##

### 进程 ###

*  进程是系统进行资源分配和调度的基本单位，是操作系统结构的基础。简单可以理解为正在运行的程序的实例，都会有一个pid作为标识。

### 多任务 ###

*  以操作系统为例，在上面可以同时运行QQ word 微信等多个程序，这就是多任务
 *  之前的单核CPU也可以运行多任务，让任务交替执行，因为CPU执行效率很快，所以对用户是无感知的
 *  真正的多任务只能在多核CPU上完成，但如果任务数量大于CPU核心数量，操作系统也会自动把很多任务轮流调度到每个核心上执行

## 创建进程 ##

### fork创建进程 ###

#### fork创建子进程 ####

*  python中的os模块封装了常见的系统调用，其中包含fork，通过fork可以轻松的创建子进程

import os 
    import time 
    
    pid = os.fork()
    print('pid is %s' % pid)
    if pid < 0:
        print('fork调用失败')
    elif pid == 0:
        # while True:
        #     print('this is 子进程')
        #     time.sleep(1)
        print('this is child process(%s) and parent is (%s)' % (os.getpid(), os.getppid()))
    else:
        print('this is parent process(%s) and my child is (%s)' % (os.getpid(), pid))
    
    print('父子进程都可以执行这里的代码')

输出结果为：

pid is 6661
    this is parent process(6660) and my child is (6661)
    父子进程都可以执行这里的代码
    pid is 0
    this is child process(6661) and parent is (6660)
    父子进程都可以执行这里的代码

*  在unix和Linux中，fork这个系统函数比较特殊，调用一次，返回两次。因为操作系统自动把当前进程复制一份，然后分别在父进程和子进程中返回
 *  当程序执行到fork时，操作系统会创建一个新的进程(子进程),然后复制父进程的所有信息到子进程中
 *  父进程和子进程都会从fork函数中得到一个返回值，在子进程中该值为0，父进程中则是子进程的ID

#### 多进程修改全局变量 ####

*  多进程中，每个进程的变量(包括全局变量)都维系一份，互不影响

import os
    import time
    
    rac = os.fork()
    num = 1
    if rac == 0:
        time.sleep(1)
        num += 1
        print('---parent num = %d' % num)
    else:
        num += 1
        time.sleep(1)
        print('---child num = %d' % num)
    
    print('---end num = %d' % num)

输出结果为：

---child num = 2 ---end num = 2 ---parent num = 2 ---end num = 2

#### 多次fork ####

*  每次fork都会创建一个子进程，并且后续代码每个进程都会执行到。

import os
    import time
    
    rac = os.fork()
    if rac == 0:
        print('---demo1---')
    else:
        print('---demo2---')
    
    rac2 = os.fork()
    if rac2 == 0:
        print('---demo3---')
    else:
        print('---demo4---')
    
    print('---end---')

输出结果为：

---demo2--- ---demo4--- ---end--- ---demo3--- ---end--- ---demo1--- ---demo4--- ---end--- ---demo3--- ---end---

*  如上所示，demo1 demo2各执行一次，demo3 demo4各执行两次，end累计执行四次

#### 父子进程执行顺序 ####

*  执行顺序没有规律，取决于操作系统的调度算法

### multiprocessing创建进程 ###

> 上面提到的fork，只能在linux unix上调用，在Windows上是无法执行的，但python作为一个跨平台的语言，自然而然也提供了一个跨平台多进程的模块–multiprocessing

#### 通过process创建进程 ####

*  Process代表一个进程对象
 *  创建子进程时，只需要在构造函数内传入目标函数以及参数，调用start方法即可
 *  join()方法是让主进程等待子进程结束后再执行后续代码

import os
    from multiprocessing import Process
    from time import sleep
    
    
    def run_proc(name):
        print('子进程运行中，name = %s，pid = %d ...' % (name, os.getpid()))
    
    
    """ 创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start方法启动 join方法可以等待子进程结束后再往下运行，通常用于进程间的同步 """
    if __name__ == '__main__':
        print('父进程 %d.' % os.getpid())
        p = Process(target=run_proc, args=('test',))
        print('子进程将要执行')
        p.start()
        p.join()
        print('子进程执行完毕')

输出结果为：

父进程 6784.
    子进程将要执行
    子进程运行中，name = test，pid = 6785 ...
    子进程执行完毕

#### 通过Process子类创建进程 ####

*  创建进程还可以使用类的方式，自定义一个类，继承Process即可

import time
    from multiprocessing import Process
    
    
    class MyProcess(Process):
        def __init__(self, time, ):
            Process.__init__(self)
            self.time = time
    
        def run(self):
            # while True:
            print('---run---')
            # time.sleep(self.time)
    
    
    if __name__ == '__main__':
        p = MyProcess(1)
        p.start()
        p.join()
        print('--main end--')

输出结果为：

---run---
    --main end--

### 进程池Pool ###

*  当创建的子进程数量不多时，可以通过Process来创建，如果要创建的子进程数量太多，那么就不适用这种方式了，需要使用进程池来解决。
 *  初始化一个Pool时，可以指定一个最大进程数，如果超过这个数，那么请求就会等待，直到池中有进程结束，才会创建新的进程来执行

from multiprocessing import Pool
    import time, os, random
    
    
    def worker(msg):
        t_start = time.time()
        print('%s 开始执行，进城号为 %d' % (msg, os.getpid()))
        # random.random()随机生成0-1的浮点数
        time.sleep(random.random() * 2)
        t_stop = time.time()
        print(msg, " 执行完毕，耗时%0.2f" % (t_stop - t_start))
    
    
    # 创建一个进程池，里面最多三个进程
    pool = Pool(3)
    for i in range(0, 10):
        # 向进程池中添加人物
        # 如果人物的数量超过进程池大小的话，会等待进程池中有空闲的时候，自动添加进去
        pool.apply_async(worker, args=(i,))
    
    print('---start---')
    # 关闭进程池 关闭后po不再接受新的请求
    pool.close()
    # 等待po中所有子进程执行完成，必须放在close语句之后
    # 如果没有join，会导致进程中的任务不会执行
    pool.join()
    print('---end---')

输出结果为

---start---
    0 开始执行，进城号为 6804
    1 开始执行，进城号为 6805
    2 开始执行，进城号为 6803
    0  执行完毕，耗时1.23
    3 开始执行，进城号为 6804
    2  执行完毕，耗时1.29
    4 开始执行，进城号为 6803
    1  执行完毕，耗时1.85
    5 开始执行，进城号为 6805
    4  执行完毕，耗时0.82
    6 开始执行，进城号为 6803
    3  执行完毕，耗时0.90
    7 开始执行，进城号为 6804
    6  执行完毕，耗时0.54
    8 开始执行，进城号为 6803
    5  执行完毕，耗时0.89
    9 开始执行，进城号为 6805
    7  执行完毕，耗时0.70
    9  执行完毕，耗时0.92
    8  执行完毕，耗时1.77
    ---end---

*  apply\_async 异步非阻塞
 *  apply 阻塞，上个进程结束才会执行下一个进程
 *  close 关闭pool 不再接收请求
 *  join 主进程阻塞，等待子进程运行结束，必须在closez以后

### 进程间通信 ###

#### 进程中的Queue ####

*  进程间有时需要互相通信，操作系统提供了多种方式进行通信，Queue就是其中的一种
 *  queue就是一个消息队列，提供了get() put() qsize() empty() full()等方法对队列进行操作
 *  下面以queue为例，创建两个进程，分别用来存储数据

from multiprocessing import Queue, Process
    import random, time
    
    
    # queue，实现多进程之间的数据传递，其实就是个消息队列
    
    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()
                print("Get value is %s" % value)
                time.sleep(random.random())
            else:
                break
    
    
    if '__main__' == __name__:
        q = Queue()
        qw = Process(target=write, args=(q,))
        qr = Process(target=read, args=(q,))
        qw.start()
        qw.join()
        qr.start()
        qr.join()
        print('---end---')
        print(random.randint(2, 4))
        print(random.random())

输出结果为：

put A to queue
    put B to queue
    put C to queue
    Get value is A
    Get value is B
    Get value is C
    ---end---
    4
    0.5260624608369131

#### 进程池中的Queue ####

*  如果使用进程池pool创建进程的话，就需要使用Manager().Queue()

from multiprocessing import Manager, Process, Pool
    import threading
    import random, time,os
    
    
    # queue，实现多进程之间的数据传递，其实就是个消息队列
    
    def write(q):
        print('---write thread is %s' % os.getpid())
        for value in ['A', 'B', 'C']:
            print("put %s to queue" % value)
            q.put(value)
    
    
    def read(q):
        print('---read thread is %s' % os.getpid())
        for i in range(q.qsize()):
            print("Get value is %s" % q.get(True))
    
    
    if '__main__' == __name__:
        print('---main thread is %s' % os.getpid())
        q = Manager().Queue()
        po = Pool()
        po.apply(write, args=(q,))
        po.apply(read, args=(q,))
        po.close()
        po.join()
        print('---end---')

输出结果为：

---main thread is 6907
    ---write thread is 6909
    put A to queue
    put B to queue
    put C to queue
    ---read thread is 6910
    Get value is A
    Get value is B
    Get value is C
    ---end---

### 总结 ###

*  本文主要介绍了进程概念 多任务 进程池 创建进程的几种方式以及进程间的通信等
 *  fork 不建议使用，不支持跨平台
 *  process  
    
    
     *  process函数形式
     *  继承Process，实现子类
 *  pool  
    
    
     *  进程池，当要创建的进程数量比较多时，建议使用该方式
 *  queue  
    
    
     *  进程间通信
     *  使用process创建的进程，使用Queue()
     *  使用进程池创建的进程通信，使用Manager().Queue()