zoukankan      html  css  js  c++  java
  • 多任务:进程

    1. 进程

    程序:例如xxx.py这是程序,是一个静态的

    进程:一个程序运行起来后,代码+用到的资源 称之为进程,它是操作系统分配资源的基本单元。

    不仅可以通过线程完成多任务,进程也是可以的。

    程序只有一个,但进程可以有多个。程序运行起来就是进程。程序不占用任何资源,进程会占用资源(CPU、内存、摄像头、键盘、光标等)

    2. 进程的状态

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

    • 就绪态:运行的条件都已经慢去,正在等在cpu执行
    • 执行态:cpu正在执行其功能
    • 等待态:等待某些条件满足,例如一个程序sleep了,此时就处于等待态

    进程的创建-multiprocessing

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

    1. 2个while循环一起执行

     1 # -*- coding:utf-8 -*-
     2 from multiprocessing import Process
     3 import time
     4 
     5 
     6 def run_proc():
     7     """子进程要执行的代码"""
     8     while True:
     9         print("----2----")
    10         time.sleep(1)
    11 
    12 
    13 if __name__=='__main__':
    14     p = Process(target=run_proc)
    15     p.start()
    16     while True:
    17         print("----1----")
    18         time.sleep(1)

    说明

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

    2、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(进程号)

    3. 进程pid

     1 # -*- coding:utf-8 -*-
     2 from multiprocessing import Process
     3 import os
     4 import time
     5 
     6 def run_proc():
     7     """子进程要执行的代码"""
     8     print('子进程运行中,pid=%d...' % os.getpid())  # os.getpid获取当前进程的进程号
     9     print('子进程将要结束...')
    10 
    11 if __name__ == '__main__':
    12     print('父进程pid: %d' % os.getpid())  # os.getpid获取当前进程的进程号
    13     p = Process(target=run_proc)
    14     p.start()

    4. 给子进程指定的函数传递参数

     1 # -*- coding:utf-8 -*-
     2 from multiprocessing import Process
     3 import os
     4 from time import sleep
     5 
     6 def run_prc(name, age, **kwargs):
     7     for i in range(10):
     8         print('子进程运行中,name= %s,age=%d ,pid=%d...' % (name, age, os.getpid()))
     9         print(kwargs)
    10         sleep(0.2)
    11 def main():
    12     p = Process(target=run_prc, args=('test',18), kwargs={"m":20})
    13     p.start()
    14 if __name__=='__main__':
    15     main()
    16     sleep(1)  # 1秒中之后,立即结束子进程
    17     p.terminate()
    18     p.join()

    运行结果:

    子进程运行中,name= test,age=18 ,pid=45097...
    {'m': 20}
    子进程运行中,name= test,age=18 ,pid=45097...
    {'m': 20}
    子进程运行中,name= test,age=18 ,pid=45097...
    {'m': 20}
    子进程运行中,name= test,age=18 ,pid=45097...
    {'m': 20}
    子进程运行中,name= test,age=18 ,pid=45097...
    {'m': 20}

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

     1 # -*- coding:utf-8 -*-
     2 from multiprocessing import Process
     3 import os
     4 import time
     5 
     6 nums = [11, 22]
     7 
     8 def work1():
     9     """子进程要执行的代码"""
    10     print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))
    11     for i in range(3):
    12         nums.append(i)
    13         time.sleep(1)
    14         print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))
    15 
    16 def work2():
    17     """子进程要执行的代码"""
    18     print("in process2 pid=%d ,nums=%s" % (os.getpid(), nums))
    19 
    20 if __name__ == '__main__':
    21     p1 = Process(target=work1)
    22     p1.start()
    23     p1.join()
    24 
    25     p2 = Process(target=work2)
    26     p2.start()

    运行结果:

    in process1 pid=11349 ,nums=[11, 22]
    in process1 pid=11349 ,nums=[11, 22, 0]
    in process1 pid=11349 ,nums=[11, 22, 0, 1]
    in process1 pid=11349 ,nums=[11, 22, 0, 1, 2]
    in process2 pid=11350 ,nums=[11, 22]

    进程间通信-Queue

    
    

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

    
    

    1. Queue的使用

    
    

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

     1 #coding=utf-8
     2 from multiprocessing import Queue
     3 q=Queue(3) #初始化一个Queue对象,最多可接收三条put消息
     4 q.put("消息1") 
     5 q.put("消息2")
     6 print(q.full())  #False
     7 q.put("消息3")
     8 print(q.full()) #True
     9 
    10 #因为消息列队已满下面的try都会抛出异常,第一个try会等待2秒后再抛出异常,第二个Try会立刻抛出异常
    11 try:
    12     q.put("消息4",True,2)
    13 except:
    14     print("消息列队已满,现有消息数量:%s"%q.qsize())
    15 
    16 try:
    17     q.put_nowait("消息4")
    18 except:
    19     print("消息列队已满,现有消息数量:%s"%q.qsize())
    20 
    21 #推荐的方式,先判断消息列队是否已满,再写入
    22 if not q.full():
    23     q.put_nowait("消息4")
    24 
    25 #读取消息时,先判断消息列队是否为空,再读取
    26 if not q.empty():
    27     for i in range(q.qsize()):
    28         print(q.get_nowait())

    运行结果: 

    False
    True
    消息列队已满,现有消息数量:3
    消息列队已满,现有消息数量:3
    消息1
    消息2
    消息3
    
    说明

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

    2. Queue实例

    我们以Queue为例,在父进程中创建两个子进程,一个往Queue里写数据,一个从Queue里读数据:

     1 from multiprocessing import Process, Queue
     2 import os, time, random
     3 
     4 # 写数据进程执行的代码:
     5 def write(q):
     6     for value in ['A', 'B', 'C']:
     7         print('Put %s to queue...' % value)
     8         q.put(value)
     9         time.sleep(random.random())
    10 
    11 # 读数据进程执行的代码:
    12 def read(q):
    13     while True:
    14         if not q.empty():
    15             value = q.get(True)
    16             print('Get %s from queue.' % value)
    17             time.sleep(random.random())
    18         else:
    19             break
    20 
    21 if __name__=='__main__':
    22     # 父进程创建Queue,并传给各个子进程:
    23     q = Queue()
    24     pw = Process(target=write, args=(q,))
    25     pr = Process(target=read, args=(q,))
    26     # 启动子进程pw,写入:
    27     pw.start()    
    28     # 等待pw结束:
    29     pw.join()
    30     # 启动子进程pr,读取:
    31     pr.start()
    32     pr.join()
    33     # pr进程里是死循环,无法等待其结束,只能强行终止:
    34     print('')
    35     print('所有数据都写入并且读完')
    
    

    进程池Pool

    
    

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

    multiprocessing.Pool常用函数解析:

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

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

     1 # -*- coding:utf-8 -*-
     2 from multiprocessing import Pool
     3 import os, time, random
     4 
     5 def worker(msg):
     6     t_start = time.time()
     7     print("%s开始执行,进程号为%d" % (msg,os.getpid()))
     8     # random.random()随机生成0~1之间的浮点数
     9     time.sleep(random.random()*2) 
    10     t_stop = time.time()
    11     print(msg,"执行完毕,耗时%0.2f" % (t_stop-t_start))
    12 
    13 po = Pool(3)  # 定义一个进程池,最大进程数3
    14 for i in range(0,10):
    15     # Pool().apply_async(要调用的目标,(传递给目标的参数元祖,))
    16     # 每次循环将会用空闲出来的子进程去调用目标
    17     po.apply_async(worker,(i,))
    18 
    19 print("----start----")
    20 po.close()  # 关闭进程池,关闭后po不再接收新的请求
    21 po.join()  # 等待po中所有子进程执行完成,必须放在close语句之后
    22 print("-----end-----")

    运行结果:

    ----start----
    0开始执行,进程号为21466
    1开始执行,进程号为21468
    2开始执行,进程号为21467
    0 执行完毕,耗时1.01
    3开始执行,进程号为21466
    2 执行完毕,耗时1.24
    4开始执行,进程号为21467
    3 执行完毕,耗时0.56
    5开始执行,进程号为21466
    1 执行完毕,耗时1.68
    6开始执行,进程号为21468
    4 执行完毕,耗时0.67
    7开始执行,进程号为21467
    5 执行完毕,耗时0.83
    8开始执行,进程号为21466
    6 执行完毕,耗时0.75
    9开始执行,进程号为21468
    7 执行完毕,耗时1.03
    8 执行完毕,耗时1.05
    9 执行完毕,耗时1.69
    -----end-----
    
    

    进程池中的Queue

    
    

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

    
    

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

    
    

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

     1 # -*- coding:utf-8 -*-
     2 
     3 # 修改import中的Queue为Manager
     4 from multiprocessing import Manager,Pool
     5 import os,time,random
     6 
     7 def reader(q):
     8     print("reader启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))
     9     for i in range(q.qsize()):
    10         print("reader从Queue获取到消息:%s" % q.get(True))
    11 
    12 def writer(q):
    13     print("writer启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))
    14     for i in "itcast":
    15         q.put(i)
    16 
    17 if __name__=="__main__":
    18     print("(%s) start" % os.getpid())
    19     q = Manager().Queue()  # 使用Manager中的Queue
    20     po = Pool()
    21     po.apply_async(writer, (q,))
    22 
    23     time.sleep(1)  # 先让上面的任务向Queue存入数据,然后再让下面的任务开始从中取数据
    24 
    25     po.apply_async(reader, (q,))
    26     po.close()
    27     po.join()
    28     print("(%s) End" % os.getpid())
    
    

    运行结果:

    
    
    (11095) start
    writer启动(11097),父进程为(11095)
    reader启动(11098),父进程为(11095)
    reader从Queue获取到消息:i
    reader从Queue获取到消息:t
    reader从Queue获取到消息:c
    reader从Queue获取到消息:a
    reader从Queue获取到消息:s
    reader从Queue获取到消息:t
    (11095) End
     
  • 相关阅读:
    1144 The Missing Number (20分)
    1145 Hashing
    1146 Topological Order (25分)
    1147 Heaps (30分)
    1148 Werewolf
    1149 Dangerous Goods Packaging (25分)
    TypeReference
    Supervisor安装与配置()二
    谷粒商城ES调用(十九)
    Found interface org.elasticsearch.common.bytes.BytesReference, but class was expected
  • 原文地址:https://www.cnblogs.com/GouQ/p/13671659.html
Copyright © 2011-2022 走看看