Python 基础之多任务 Process 进程知识点的简单整理，以及对应的使用（进程队列，进程池，进程池队列等）

偏执的太偏执、 2022-10-29 07:33 125阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70][]

# Python 基础 之 多任务 Process 进程知识点的简单整理，以及对应的使用（进程队列，进程池，进程池队列等） #

**目录**

Python 基础 之 多任务 Process 进程知识点的简单整理，以及对应的使用（进程队列，进程池，进程池队列等）

一、简单介绍

二、进程的相关概念

三、进程的基本使用

四、带参数的进程使用

五、进程间不共享全局变量

六、进程间通讯队列 - Queue

七、进程池Pool 的使用

八、进程池 Pool 中的队列 Queue 使用

--------------------

## 一、简单介绍 ##

Python是一种跨平台的计算机程序设计语言。是一种面向对象的动态类型语言，最初被设计用于编写自动化脚本(shell)，随着版本的不断更新和语言新功能的添加，越多被用于独立的、大型项目的开发。Python是一种解释型脚本语言，可以应用于以下领域： Web 和 Internet 开发、科学计算和统计、人工智能、教育、桌面界面开发、软件开发、后端开发、网络爬虫。

本节简单介绍，简单的整理一下 python  中多任务 Process 的一些知识点，和简单的案例说明。

***注意：子进程的报错不会显示在控制台，或者说子进程报错不会给提醒提示！！！***

## 二、进程的相关概念 ##

> **进程、状态、进程的创建、进程和线程对比、进程间通讯Queue、进程池Pool、进程池中的Queue、**
> 
> 1）进程
> 
> 程序：例如xxx.py这是程序，是一个静态的
> 
> 进程：一个程序运行起来后，代码+用到的资源 称之为进程，它是操作系统分配资源的基本单元。
> 
> 不仅可以通过线程完成多任务，进程也是可以的
> 
>  
> 
> 2）进程的状态
> 
> 工作中，任务数往往大于cpu的核数，即一定有一些任务正在执行，而另外一些任务在等待cpu进行执行，因此导致了有了不同的状态
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 1][]
> 
>  *  就绪态：运行的条件都已经慢去，正在等在cpu执行
>  *  执行态：cpu正在执行其功能
>  *  等待态：等待某些条件满足，例如一个程序sleep了，此时就处于等待态
> 
>  
> 
> 3）进程的创建-multiprocessing
> 
> multiprocessing模块就是跨平台版本的多进程模块，提供了一个Process类来代表一个进程对象，这个对象可以理解为是一个独立的进程，可以执行另外的事情。
> 
> 创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动。
> 
> ***Process语法结构如下：***
> 
> Process(\[group \[, target \[, name \[, args \[, kwargs\]\]\]\]\])
> 
>  *  target：如果传递了函数的引用，可以任务这个子进程就执行这里的代码
>  *  args：给target指定的函数传递的参数，以元组的方式传递
>  *  kwargs：给target指定的函数传递命名参数
>  *  name：给进程设定一个名字，可以不设定
>  *  group：指定进程组，大多数情况下用不到
> 
> Process创建的实例对象的常用方法：
> 
>  *  start()：启动子进程实例（创建子进程）
>  *  is\_alive()：判断进程子进程是否还在活着
>  *  join(\[timeout\])：是否等待子进程执行结束，或等待多少秒
>  *  terminate()：不管任务是否完成，立即终止子进程
> 
> Process创建的实例对象的常用属性：
> 
>  *  name：当前进程的别名，默认为Process-N，N为从1开始递增的整数
>  *  pid：当前进程的pid（进程号）
> 
> **注意：进程间不同享全局变量**
> 
>

> 4）进程、线程对比
> 
> 功能
> 
>  *  进程，能够完成多任务，比如 在一台电脑上能够同时运行多个QQ
>  *  线程，能够完成多任务，比如 一个QQ中的多个聊天窗口
> 
> 定义的不同
> 
>  *  进程是系统进行资源分配和调度的一个独立单位.
>  *  线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈),但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源.
> 
> 区别
> 
>  *  一个程序至少有一个进程,一个进程至少有一个线程.
>  *  线程的划分尺度小于进程(资源比进程少)，使得多线程程序的并发性高。
>  *  进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的运行效率
>  *  线线程不能够独立执行，必须依存在进程中
>  *  可以将进程理解为工厂中的一条流水线，而其中的线程就是这个流水线上的工
> 
> 优缺点
> 
> 线程和进程在使用上各有优缺点：线程执行开销小，但不利于资源的管理和保护；而进程正相反。
> 
>  
> 
> 5）进程间通信-Queue
> 
> Process之间有时需要通信，操作系统提供了很多机制来实现进程间的通信。
> 
> Queue的使用
> 
> 可以使用multiprocessing模块的Queue实现多进程之间的数据传递，Queue本身是一个消息列队程序。
> 
> 初始化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)；
> 
>  
> 
> 6）进程池Pool
> 
> 当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。
> 
> 初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务。
> 
> multiprocessing.Pool常用函数解析：
> 
>  *  apply\_async(func\[, args\[, kwds\]\]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
>  *  close()：关闭Pool，使其不再接受新的任务；
>  *  terminate()：不管任务是否完成，立即终止；
>  *  join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；
> 
>  
> 
> 7）进程池中的Queue
> 
> 如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否则会得到一条如下的错误信息：
> 
> RuntimeError: Queue objects should only be shared between processes through inheritance.

## 三、进程的基本使用 ##

进程的创建-multiprocessing

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

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

***Process语法结构如下：***

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

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

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

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

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

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

**注意：进程间不同享全局变量**

效果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 2][]

代码：

import multiprocessing
    import os
    
    
    def run_process():
        # 获取打印当前进程 id
        print("子进程运行中，pid=%d ..." % os.getpid())
        pass
    
    
    def main():
        # 获取打印当前进程 id
        print("父进程pid：%d ..." % os.getpid())
    
        # 创建子进程，并执行
        p = multiprocessing.Process(target=run_process)
        p.start()
        pass
    
    
    if __name__ == '__main__':
        main()

## 四、带参数的进程使用 ##

效果：可以吧参数传给子进程

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 3][]

代码：

import multiprocessing
    import os
    import time
    
    
    def run_process(name, age, **kwargs):
        """
        给子进程传递传输
        :param name:
        :param age:
        :param kwargs:
        :return:
        """
    
        for i in range(0, 10):
            print("子进程运行中，name = %s, age = %d, pid = %d" % (name, age, os.getpid()))
            print(kwargs)
            time.sleep(0.3)
    
        pass
    
    
    def main():
        # 获取打印当前进程 id
        print("父进程pid：%d ..." % os.getpid())
    
        p = multiprocessing.Process(target=run_process, args=("Jane", 18), kwargs={"other": 0})
        p.start()
        time.sleep(1)
    
        # 中断子进程
        p.terminate()
        # 等待子进程结束
        p.join()
    
        pass
    
    
    if __name__ == '__main__':
        main()

## 五、进程间不共享全局变量 ##

效果：全局变量，每个进程独立使用自己的那份

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 4][]

代码：

import multiprocessing
    import os
    
    
    nums = [11, 22]
    
    
    def run_process_one():
        # 获取打印当前进程 id
        print("子进程运行中，pid=%d , nums=%s ..." % (os.getpid(), nums))
        for i in range(4):
            nums.append(i)
            print("子进程运行中，pid=%d , nums=%s ..." % (os.getpid(), nums))
    
    
    def run_process_two():
        # 获取打印当前进程 id
        print("子进程运行中，pid=%d , nums=%s ..." % (os.getpid(), nums))
        for i in range(4):
            nums.append(i)
            print("子进程运行中，pid=%d , nums=%s ..." % (os.getpid(), nums))
    
    
    def main():
    
        p1 = multiprocessing.Process(target=run_process_one)
        p1.start()
    
        # 等待进程 1 执行完
        p1.join()
    
        p2 = multiprocessing.Process(target=run_process_two)
        p2.start()
    
        pass
    
    
    if __name__ == '__main__':
        main()

## 六、进程间通讯队列 - Queue ##

Process之间有时需要通信，操作系统提供了很多机制来实现进程间的通信。

Queue的使用

可以使用multiprocessing模块的Queue实现多进程之间的数据传递，Queue本身是一个消息列队程序。

初始化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)；

效果：刚好解决进程间不共享全局变量的特点，使用Queue 进行进程间通讯

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 5][]

代码：

import multiprocessing
    import time, random
    
    
    def run_process_put_data(queue):
        """把数据入队队列"""
        for value in ['e', 'r', 'j']:
            print('Put %s to queue ...' % value)
            queue.put(value)
            time.sleep(random.random())
        pass
    
    
    def run_process_get_data(queue):
        """把数据出队队列"""
        while True:
            if not queue.empty():
                value = queue.get()
                print('Get %s from queue ...' % value)
                time.sleep(random.random())
            else:
                break
        pass
    
    
    def main():
        # 在父进程创建一个 Queue, 并传给各个子进程
        q = multiprocessing.Queue()
    
        # 创建子进程，写入队列数据
        p_put = multiprocessing.Process(target=run_process_put_data, args=(q, ))
        p_put.start()
        p_put.join()
    
        # 创建子进程，读取队列数据
        p_get = multiprocessing.Process(target=run_process_get_data, args=(q,))
        p_get.start()
        p_get.join()
    
        print("队列数据写入并且读完...")
    
        pass
    
    
    if __name__ == '__main__':
        main()

## 七、进程池Pool 的使用 ##

进程池Pool

当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。

初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务。

multiprocessing.Pool常用函数解析：

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

效果：可以复用进程池空闲的进程，保证不必要的进程开销创建

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 6][]

代码：

import multiprocessing
    import time,random,os
    
    
    def run_process(msg):
        t_start = time.time()
        print("%s 序号进程开始执行，进程号为%d" % (msg, os.getpid()))
    
        time.sleep(random.random())
    
        t_stop = time.time()
        print(msg, " 执行完，耗时 %.2f" % (t_stop-t_start))
    
    
    def main():
        # 定义一个进程池，允许最大同时运行的进程数为 3
        pool = multiprocessing.Pool(3)
    
        # 向进程池中添加 10 个进程任务(有空闲进程的时候会自动自行进程)
        for i in range(0, 10):
            # 向进程池中添加进程任务
            pool.apply_async(run_process, (i, ))
    
        pass
    
        print("====进程任务开始====")
        # 关闭进程池，不再接受新的进程任务
        pool.close()
        # 主进程等待子进程执行任务完成
        pool.join()
        print("====进程任务结束====")
    
    
    if __name__ == '__main__':
        main()

## 八、进程池 Pool 中的队列 Queue 使用 ##

进程池中的Queue

如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否则会得到一条如下的错误信息：

RuntimeError: Queue objects should only be shared between processes through inheritance.

效果：可以冲进程池中得到各个进程入队的数据

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 7][]

代码：

import multiprocessing
    import time, os
    
    
    def run_process_put_data(queue):
        print("进程 %d put data, 父进程是 %d  " % (os.getpid(), os.getppid()))
        for i in "ILoveYou":
            print("put data 到队列中 ：%s" % i)
            queue.put(i)
    
    
    def run_process_get_data(queue):
        print("进程 %d get data, 父进程是 %d  " % (os.getpid(), os.getppid()))
        for i in range(queue.qsize()):
            value = queue.get()
            print("get data 到队列中 ：%s" % value)
    
    
    def main():
        # 获取打印当前进程 id
        print("父进程pid：%d 开始..." % os.getpid())
    
        # 进程池中的队列
        queue = multiprocessing.Manager().Queue()
    
        # 创建进程池
        pool = multiprocessing.Pool()
    
        # 添加进程任务，数据入队
        pool.apply_async(run_process_put_data, (queue, ))
    
        # 等待进程数据入队
        time.sleep(1)
    
        # 添加进程任务，数据出队
        pool.apply_async(run_process_get_data, (queue, ))
    
        # 关闭进程池任务添加，等到子进程结束
        pool.close()
        pool.join()
    
        # 获取打印当前进程 id
        print("父进程pid：%d 结束..." % os.getpid())
    
        pass
    
    
    if __name__ == '__main__':
        main()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 8][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70]: /images/20221024/19ff42d126714bec9845f1c1dd9527f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 1]: /images/20221024/a8168d7797d14be896ec8336b3afc033.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 2]: /images/20221024/551cf379f24e4ba0a4a6da2dc5c186d4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 3]: /images/20221024/b87d860cae7f4ab891fde916d038f2c5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 4]: /images/20221024/ce84f517d3614793a907f4603bbbfe5f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 5]: /images/20221024/6d6b34dd50a4473489fc3aa40126e53d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 6]: /images/20221024/f62f3de30f534ca29d7325ae47e13be9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 7]: /images/20221024/d6dbd4f7366e4494a08d0ada98e5c9b0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNjEyODA_size_16_color_FFFFFF_t_70 8]: /images/20221024/971a2c9bea9c463c93b2667a53c6f4b5.png