多任务--进程

Posted 2020-11-22 xiaoshust

目录

进程以及状态

进程的创建--multiprocessing


进程、线程对比

进程间通信--Queue

一、进程的创建--multiprocessing

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

　　1.2个while循环一起执行

　　　　# -*- coding:utf-8 -*-

　　　　from multiprocessing import Process

　　　　import time

 

　　　　def run_proc():

　　　　　　print(‘---2---‘)

　　　　　　time.sleep(1)

 

　　　　if __name__ == ‘__main__‘:

　　　　　　p = Process(target=run_proc)

　　　　　　p.start()

　　　　　　while True:

　　　　　　　　print(‘---1---‘)

　　　　　　　　time.sleep(1)

 

　　说明

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

　　

　　2.进程pid

　　　　# -*- coding:utf-8 -*-

　　　　from multiprocessing import Process

　　　　import os

　　　　import time

　　　　

　　　　def run_proc():

　　　　　　‘‘‘子进程要执行的代码‘‘‘

　　　　　　print(‘子进程运行中，pid=%d...‘ % os.getpid())　　#os.getpid获取当前进程的进程号

　　　　　　print(‘子进程将要结束‘)

 

　　　　if __name__ = ‘__main__‘:

　　　　　　print(‘父进程pid: %d‘ % os.getpid())　　#os.getpid获取当前进程的进程号

　　　　　　p = process(target=run_proc)

　　　　　　p.start()

　　　　

　　　　3.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.给子进程指定的函数传递参数

　　　　　　

　　　　　　# -*- coding:utf-8 -*-
　　　　　　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()

　　　　
　　　　5.进程间不同享全局变量
　　　　　　

　　　　# -*- coding:utf-8 -*-
　　　　from multiprocessing import Process
　　　　import os
　　　　import time

　　　　nums = [11, 22]

　　　　def work1():
    　　　　"""子进程要执行的代码"""
    　　　　print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))
    　　　　for i in range(3):
       　　　　 nums.append(i)
        　　　　time.sleep(1)
       　　　　 print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))

　　　　def work2():
    　　　　"""子进程要执行的代码"""
    　　　　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()

三、进程、线程对比

　　功能

• • 　　进程，能够完成多任务，比如 在一台电脑上能够同时运行多个QQ
  • 　　线程，能够完成多任务，比如 一个QQ中的多个聊天窗口

　　定义的不同

　　　　进程是系统进行资源分配和调度的一个独立单位

　　　　线程是进程的一个实体，是CPU调度和分派的基本单位，它是比进程更小的能独立运行的基本单位，线程自己基本上不拥有系统资源，只拥有一点运行中必不可少的资源（如

　　　　程序计数器，一组寄存器和栈），但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源

 

　　区别

　　　　一个程序至少有一个进程，一个进程至少有一个线程

　　　　线程的划分尺度小于进程（资源比进程少），使的多线程程序的并发性高

　　　　进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的于心效率

　　　　

 

　　　　线程不能独立运行，必须依存在进程中

　　　　可以将进程理解为工厂中国呢的一条流水线，而其中的线程就是这个流水线上的工人

　　　　

 

 

　　　　优缺点

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

 

四、进程间通信--Queue

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

　　1.Queue的使用

　　　　可以使用multiprocess模块的Queue实现多进程之间的数据传递，Queue本身是一个消息列队程序，首先用一个小实例来演示以下Queue的工作原理

　　　　

　　　　#coding=utf-8
　　　　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秒，若还没有空间，则抛出“Queue.put”异常

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

　　　　　　　　Queue.put_nowait(item):相当Queue.put(item, False)

　　2.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()
    　　# pr进程里是死循环，无法等待其结束，只能强行终止:
    　　print(‘‘)
    　　print(‘所有数据都写入并且读完‘)

 

五、进程池Pool

　　当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态的生成多个进程，但如果是上百甚至上千个目标，手动的去创建

　　进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法

　　初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程来执行该请求；但如果池中但进程数

　　已经达到指定但最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务，请看下面的实例

　　# -*- coding:utf-8 -*-
　　from multiprocessing import Pool
　　import os, time, 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-----")


 

 　　multiprocessing.pool常用函数解析

　　　　apply_async(func[, args[, kwargs]]):使用非阻塞方式调用func（并行执行，阻塞方式必须等待上一个进程推出才能执行下一个进程），args为传递给

　　　　func的参数列表，kwargs为传递给func的关键字参数列表

　　　　close（）：关闭Pool，使其不再接受新的任务

　　　　terminate（）：不管任务是否完成，立即终止

　　　　join（）：主进程阻塞，等待子进程的推出，必须在close或terminate之后使用

 

　　进程池中的Queue

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

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

　　　　下面的实例演示了进程池中的进程如何通信

　　　　

　　# -*- coding:utf-8 -*-

　　# 修改import中的Queue为Manager
　　from multiprocessing import Manager,Pool
　　import os,time,random

　　def reader(q):
    　　print("reader启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))
    　　for i in range(q.qsize()):
        　　print("reader从Queue获取到消息：%s" % q.get(True))

　　def writer(q):
    　　print("writer启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))
    　　for i in "xiaoshust":
        　　q.put(i)

　　if __name__=="__main__":
    　　print("(%s) start" % os.getpid())
    　　q = Manager().Queue()  # 使用Manager中的Queue
    　　po = Pool()
    　　po.apply_async(writer, (q,))

    　　time.sleep(1)  # 先让上面的任务向Queue存入数据，然后再让下面的任务开始从中取数据

    　　po.apply_async(reader, (q,))
    　　po.close()
    　　po.join()
    　　print("(%s) End" % os.getpid())