img

FXHao

脱发之旅~~

FXHao

个人站

欢迎来到我的小世界~哈哈哈~

多任务之进程

| 阅读：次

多任务之进程

  • 一个程序运行起来后，代码+用到的资源 称之为进程，它是操作系统分配资源的基本单元，也可以实现多任务。

进程与线程

  • 进程是系统进行资源分配和调度的一个独立单位.
  • 线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈),但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源

  • 一个程序至少有一个进程,一个进程至少有一个线程.
  • 线程的划分尺度小于进程(资源比进程少)，使得多线程程序的并发性高。
  • 线线程不能够独立执行，必须依存在进程中

优缺点

  • 线程和进程在使用上各有优缺点：线程执行开销小，但不利于资源的管理和保护；而进程正相反。

进程的状态

  • 工作中，任务数往往大于 CPU 的核数，即一定有一些任务正在执行，而另外一些任务在等待cpu进行执行，因此导致了有了不同的状态

进程的创建

  • 与线程类似，只需要将模块 threading =====> multiprocessing 然后调用该模块中的 Process 创建实例，在用 start() 启动
import multiprocessing
import time

def test1():
    while True:
        print("1--------")
        time.sleep(1)

def test2():
    while True:
        print("--------2")
        time.sleep(1)

def main():
    p1 = multiprocessing.Process(target=test1)  # 创建进程对象
    p1.start()  # 启动进程
    p2 = multiprocessing.Process(target=test2)
    p2.start()

if __name__ == '__main__':
    main()

  • 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（进程号）

进程之间的通信

  • 首先，进程之间是不共享全局变量的，它们之间的通信是靠 Queue来实现的，Queue是一个队列(先进先出)

Queue的使用

import multiprocessing

def download_form_web(q):
    """下载数据"""
    # 模拟从网上下载的数据
    data = [11, 22, 33, 44]
    # 向队列中写入数据
    for temp in data:
        q.put(temp)
    print("下载器已经下载完了数据并且存入到队列q中。。。。")

def analysis_data(q):
    """数据处理"""
    waiting_analysis_data = list()
    # 从队列中获取数据
    while True:
        data = q.get()
        waiting_analysis_data.append(data)
        if q.empty():
            break
    # 模拟数据处理
    print(waiting_analysis_data)

def main():
    # 1.创建一个队列
    q = multiprocessing.Queue()
    # 2.创建多个进程，将队列的引用当做实参传递到里面
    p1 = multiprocessing.Process(target=download_form_web, args=(q,))
    p1.start()
    p2 = multiprocessing.Process(target=analysis_data, args=(q,))
    p2.start()

if __name__ == '__main__':
    main()

Queue说明

  • q=Queue(n) ： 初始化 Queue()对象，n为接收消息的最大数量，不写的话就默认系统最大
  • Queue.qsize()：返回当前队列包含的消息数量
  • Queue.empty()：如果队列为空，返回True，反之False
  • Queue.full()：如果队列满了，返回True,反之False
  • Queue.get([block[, timeout]])：获取队列中的一条消息，然后将其从列队中移除，block默认值为True

进程池Pool

  • 当需要创建任务目标有很多时，逐个创建对应的子进程就很不现实，工作量非常大，这时就可以用multiprocessing模块提供的Pool方法
  • 创建一个 Pool，指定一个最大进程数，若有新的请求传到Pool时，如果Pool没满，则就创建进程，若Pool已满，那么该请求就会等待Pool中某个进程结束，在创建
from multiprocessing import Pool
import time
import os
import 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))

po = Pool(3)  # 定义一个进程池，最大进程数3
for i in range(0, 10):
    # Pool().apply_async(要调用的目标,(传递给目标的参数元祖,))
    # 每次循环将会用空闲出来的子进程去调用目标
    po.apply_async(worker, (i,))

print("----start----")
po.close()  # 关闭进程池，关闭后po不再接收新的请求
po.join()  # 等待po中所有子进程执行完成，必须放在close语句之后
print("----end----")
  • 注：如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()

multiprocessing.Pool常用函数解析：

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

案例——文件夹copy器（多进程版）

import multiprocessing
import os
import time
import random

def copy_file(queue, file_name, source_folder_name,  dest_folder_name):
    """copy文件到指定的路径"""
    f_read = open(source_folder_name + "/" + file_name, "rb")
    f_write = open(dest_folder_name + "/" + file_name, "wb")
    while True:
        time.sleep(random.random())
        content = f_read.read(1024)
        if content:
            f_write.write(content)
        else:
            break
    f_read.close()
    f_write.close()
    
    queue.put(file_name)  # 发送已经拷贝完毕的文件名字

def main():
    # 获取要复制的文件夹
    source_folder_name = input("请输入要复制文件夹名字:")
    # 整理目标文件夹
    dest_folder_name = source_folder_name + "[副本]"
    # 创建目标文件夹
    try:
        os.mkdir(dest_folder_name)
    except:
        pass  # 如果文件夹已经存在，那么创建会失败

    # 获取这个文件夹中所有的普通文件名
    file_names = os.listdir(source_folder_name)
    # 创建Queue
    queue = multiprocessing.Manager().Queue()
    # 创建进程池
    pool = multiprocessing.Pool(5)
    for file_name in file_names:
        # 向进程池中添加任务
        pool.apply_async(copy_file, args=(queue, file_name, source_folder_name, dest_folder_name))
    # 主进程显示进度
    pool.close()

    all_file_num = len(file_names)
    while True:
        file_name = queue.get()
        if file_name in file_names:
            file_names.remove(file_name)

        copy_rate = (all_file_num-len(file_names))*100/all_file_num
        print("\r%.2f...(%s)" % (copy_rate, file_name) + " "*50, end="")
        if copy_rate >= 100:
            break
    print()

if __name__ == "__main__":
    main()

END…