20python的多进程 多进程使用queue共享数据 自定义进程类

一、前情知识了解

二、python的多进程

三、python 多进程之间如何实现数据共享

四、自定义进程类

一、前情知识了解

1.什么是程序和系统

一个系统一般由多个程序构成。如：win10系统由进程管理程序、内存管理程序、n个驱动程序等构成

2.操作系统的作用：与硬件交互

3.任务：打开某个软件。浏览网页、下载电影。。。每个应用呈现被成为一个任务

4.单任务：只能有一个任务在进行（DOS系统）

5.多任务：可以有多个任务同时进行（WIN10）

6.CPU时间片：程序的执行依赖CPU，CPU把自己进行时间上的分割，每个程序轮流占用CPU。时间片一般是切换时间的100倍

7.分时操作系统：以时间片轮转的方式，让一台计算机为多个终端服务

8.进程：应用程序的动态执行过程

  进程的状态：运行、等待、就绪

  正在运行的软件，就是正在运行的进程。（一个软件只有运行，才能生成进程，可生成1-n个进程。不运行就是个软件，就不能叫做进程）

  进程的执行过程：启动--》软件代码要加载到内存--》CPU为进程建立档案（PID：进程的唯一标识）--》等CPU的时间片--》执行--》退出

9.单核：同一时间，只能有一个程序在使用CPU

  比如说多个程序放在队列中，按照某种算法让不同的程序轮流执行，0.01us允许A程序，0.02us允许B程序

二、python的多进程（多个进程并发执行）

0.通过使用子进程而非线程有效地绕过了 全局解释器锁GIL（global interpreter lock）。 因此，multiprocessing 模块允许程序员充分利用给定机器上的多个处理器。 它在 Unix 和 Windows 上均可运行

1. 在python中，执行一个python脚本，就会创建一个进程

•     python中获取当前进程的id: os.getpid()
•     手动终止进程：taskkill /f /PID 89888

2. python中如何创建多进程

• 标准库 ：multiprocessing库中Process

         from multiprocessing import Process

3. 如何创建子进程

    在主进程中通过Process类创建的进程对象就是子进程

举个栗子：

import logging
import os
import time
from multiprocessing import Process

logging.basicConfig(level=logging.INFO)


class MyClass:
    """Process 1"""

    def __init__(self, *, name: str, age: int) -> None:
        """Init"""

        self.name = name
        self.age = age
        self.logger = logging.getLogger("MyProcess")

    def info(self, address: str) -> None:
        """Info"""

        self.logger.info("This is info...")
        basic_info = (self.name, self.age)
        for _ in range(2):
            time.sleep(1)
            print(f"Basic info is {basic_info}, address{address}.")

    def sing(self) -> None:
        """Sing"""

        self.logger.info("This is sing...")
        for _ in range(3):
            time.sleep(1)
            print(f"{self.name} is singing《reality》now.")


def validation_daemon():
    """Validate daemon"""

    my_process = MyClass(name="zhangsan", age=18)
    start_time = time.time()
    # Creat process
    # daemon=False, 主进程要等待子进程,daemon=True，主进程结束，程序就结束了，不会等待子进程
    # 这里主进程运行了0.003s后，还会继续等待子进程
    process_1 = Process(target=my_process.info, args=("beijing road",), daemon=False)
    process_2 = Process(target=my_process.sing, daemon=False)
    print(f"This is Process, pid is {os.getpid()}")

    # Start
    process_1.start()
    process_2.start()
    end_time = time.time()
    print(f"Run time is {end_time - start_time}")  # 主进程最后一行，运行完主进程结束，但是是否等待子进程，取决daemon参数设置


def validation_join():
    """Validate join"""

    my_process = MyClass(name="LiSi", age=20)
    start_time = time.time()
    # Creat process
    process_1 = Process(target=my_process.info, args=("beijing road",), daemon=True)
    process_2 = Process(target=my_process.sing, daemon=True)
    print(f"This is Process, pid is {os.getpid()}")

    # Start
    process_1.start()
    process_2.start()
    # join，主进程等待子进程结束后，再执行后面的代码，那么主进程的运行时间变长（包含等待子进程运行的时间）
    process_1.join()
    process_2.join()
    end_time = time.time()
    print(f"Run time is {end_time - start_time}")

def single_process():
    """Single process """

    my_process = MyClass(name="LiSi", age=20)
    start_time = time.time()
    my_process.info(address="beijin road")
    my_process.sing()
    end_time = time.time()
    print(f"Run time is {end_time - start_time}")


if __name__ == '__main__':
    validation_daemon()
    validation_join()
    single_process()  # 单进程执行效率低，时间长

总结：

1.启动进程：process.start()

2.Process 的参数daemon(daemon:守护进程；后台程序)

• daemon=True 表示主进程执行完，不会等待该子进程
• daemon=False 表示主进程执行完，会等待该子进程执行完毕 再退出

 3.Process的参数join

• p1.join()表示：主进程会等待子进程p1执行完，再执行剩下的代码

三、python 多进程之间如何实现数据共享

1.首先来看一下，多进程之间能否直接操作全局变量进行数据共享？

思路：

• 创建一个全局变量
• 创建2个子进程，分别对这个全局变量进行操作
• 查看全局变量值 是否发生变化（2个子进程对该全局变量的操作是否成功？）

"""多进程的全局变量"""



from multiprocessing import Process

CONSTANT = 10


def funtion_1():
    """Funtion 1"""

    global CONSTANT
    CONSTANT += 1
    print(f"Funtion_1 constant is {CONSTANT}")


def funtion_2():
    """Function 2"""
    global CONSTANT
    CONSTANT += 2
    print(f"Funtion_2 constant is {CONSTANT}")


def funtion_test():
    """Test"""
    f1_process = Process(target=funtion_1)
    f2_process = Process(target=funtion_2)
    f1_process.start()
    f2_process.start()


if __name__ == '__main__':
    funtion_test()
    print(f"The main process constant is {CONSTANT}")  #10 ，因为多个进程之间的资源是不可分享的，因为各个进程是独立占用内存的

结果：全局变量没有发生变化。

结论：

每创建一个进程，CPU都会单独分配资源；所以呢，多个进程是不会共享全局变量的

2.如何实现多进程之间数据共享----多个进程共享主进程中创建的队列

方法：multiprocessing 库中的Queue模块，即用 队列queue共享数据

思路：

• 主进程创建队列
• 然后把队列作为参数传到各个子进程中
• 各子进程对队列中的数据进行操作

队列的操作：

• queue.get() 表示获取队列中的数据
• queue.put(X) 表示将数据X存入队列中

实现如下：

"""Queue"""

# 多进程中如何使用队列queue共享数据？----多个进程共享主进程中创建的队列
# 方法：
# 主进程创建队列，然后把队列作为参数传到各个子进程中即可

from multiprocessing import Process, Queue


def fun1(queue):
    """Fun 1
    :param queue:  Queue
    """
    value = queue.get()
    value += 1
    queue.put(value)
    print(f"Fun 1 value is {value}")


def fun2(queue):
    """Fun 2
    :param queue:  Queue
    """
    value = queue.get()
    value += 2
    queue.put(value)
    print(f"Fun 2 value is {value}")


def funtion_multi_process():
    queue = Queue()
    queue.put(0)
    f1_process = Process(target=fun1, args=(queue,), daemon=False)
    f2_process = Process(target=fun2, args=(queue,), daemon=False)
    f1_process.start()
    f2_process.start()
    f1_process.join() # 这里使用join()进行阻塞主进程的话，上面的daemon参数就可以不设置了
    f2_process.join()
    print(f"Main process value is {queue.get()}")


if __name__ == '__main__':
    funtion_multi_process()

四、自定义进程类

1.自定义类继承multiprocessing的Process类

"""自定义进程类"""

import multiprocessing


class MyMultiProcess(multiprocessing.Process):
    """Class for Multi Process."""

    def __init__(self, user: str):
        """Init"""
        super().__init__()
        self.user = user

    def run(self):
        num = 10
        for _ in range(5):
            num += 1
        print(f"{self.user}'s number is {num}")


if __name__ == '__main__':
    # Creat an object
    p1 = MyMultiProcess("zhangsan")
    p2 = MyMultiProcess("lisi")
    p1.start()
    p2.start()