zoukankan      html  css  js  c++  java

  • 进程

    知识内容：

    1.进程概念

    2.创建进程

    3.守护进程

    4.进程同步

    5.进程通信

    6.进程间数据共享

    7.进程池

    参考：https://www.cnblogs.com/Eva-J/articles/8253549.html

    1.进程概念

    关于进程的概念详细看此：https://www.cnblogs.com/wyb666/p/9636270.html

    multiprocess：

    multiprocess不是一个模块而是python中一个操作、管理进程的包。 之所以叫multi是取自multiple的多功能的意思，在这个包中几乎包含了和进程有关的所有子模块。

    multiprocess分为以下几部分：

    • 创建进程部分：multiprocess.Process
    • 进程同步部分：multiprocess.Lock、multiprocess.Semaphore、multiprocess.Event
    • 进程间通信：multiprocess.Queue、multiprocess.Pipe
    • 进程之间数据共享：multiprocess.Manager
    • 进程池部分：multiprocess.Pool

    2.创建进程

    (1)Process

     1 Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）
 2 
 3 强调：
 4 1. 需要使用关键字的方式来指定参数
 5 2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号
 6 
 7 参数介绍：
 8 group参数未使用，值始终为None
 9 target表示调用对象，即子进程要执行的任务
10 args表示调用对象的位置参数元组，args=(1,2,'xxx',)
11 kwargs表示调用对象的字典,kwargs={'name':'xxx','age':18}
12 name为子进程的名称

    方法介绍(下面的p是Process对象，也就是进程对象)：

    1 p.start()：启动进程，并调用该子进程中的p.run() 
2 p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法  
3 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
4 p.is_alive():如果p仍然运行，返回True
5 p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程

    属性介绍：

    1 p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
2 p.name:进程的名称
3 p.pid：进程的pid
4 p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
5 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）

    (2)创建进程实例

     1 import os
 2 from multiprocessing import Process
 3 
 4 def func(args1, args2):
 5     print("子进程开始----------------------------")
 6     print("子进程: ", os.getpid())
 7     print("子进程的父进程: ", os.getppid())
 8     print("args: ", args1, args2)
 9     print(666)
10 
11 if __name__ == '__main__':      # main只在Windows上才需要
12     # 注册:
13     # p是一个进程对象
14     p = Process(target=func, args=("参数1", "参数2"))
15     # 启动进程:
16     # 注意不是立即执行!
17     p.start()
18     print("父进程开始----------------------------")
19     print("父进程: ", os.getpid())
20     print("父进程的父进程: ", os.getppid())
21     print('*' * 10)
     1 # join方法
 2 import time
 3 from multiprocessing import Process
 4 
 5 def func(arg1,arg2):
 6     print('*'*arg1)
 7     time.sleep(5)
 8     print('*'*arg2)
 9 
10 if __name__ == '__main__':
11     p = Process(target=func,args=(10,20))
12     p.start()
13     print('hahahaha')
14     p.join()     # 感知一个子进程的结束,将异步的程序改为同步
15     print('====== : 运行完了')
     1 # 同时创建多个进程
 2 from multiprocessing import Process
 3 
 4 def func(arg1, arg2):
 5     print('*' * arg1)
 6     print('*' * arg2)
 7 
 8 if __name__ == '__main__':
 9     p_lst = []
10     for i in range(10):
11         p = Process(target=func, args=(10 * i, 20 * i))
12         p_lst.append(p)
13         p.start()
14     for p in p_lst:
15         p.join()   # 之前的所有进程必须在这里都执行完才能执行下面的代码
16     print("运行完成了")

    还可以这样创建子进程：

     1 from multiprocessing import Process
 2 
 3 class MyProcess(Process):
 4     def __init__(self, arg1, arg2):
 5         super().__init__()
 6         self.arg1 = arg1
 7         self.arg2 = arg2
 8 
 9     def run(self):
10         print(self.pid)
11         print(self.name)
12         print(self.arg1)
13         print(self.arg2)
14 
15 if __name__ == '__main__':
16     p1 = MyProcess(1, 2)
17     p1.start()
18     p2 = MyProcess(3, 4)
19     p2.start()
20 
21 # 自定义类 继承Process类
22 # 必须实现一个run方法(就是上面的func函数),run方法中是在子进程中执行的代码

    (3)多进程写文件

     1 import os
 2 from multiprocessing import Process
 3 
 4 def func(filename, content):
 5     with open(filename, 'w') as f:
 6         f.write(content*10*'*')
 7 
 8 if __name__ == '__main__':
 9     p_lst = []
10     for i in range(10):
11         p = Process(target=func, args=('info%s' % i, i))
12         p_lst.append(p)
13         p.start()
14     for p in p_lst:
15         p.join()   # 之前的所有进程必须在这里都执行完才能执行下面的代码
16     now_path = os.getcwd()
17     # print(now_path)
18     # 展示写入文件之后文件夹中的所有的文件名
19     print([i for i in os.walk(now_path)])

    注意：

    在Windows操作系统中由于没有fork(linux操作系统中创建进程的机制)，在创建子进程的时候会自动 import 启动它的这个文件，而在 import 的时候又执行了整个文件。

    因此如果将process()直接写在文件中就会无限递归创建子进程报错。所以必须把创建子进程的部分使用if __name__ ==‘__main__’ 判断保护起来，import 的时候 ，就不会递归运行了。

    3.守护进程

    (1)什么是守护进程

    守护进程会随着主进程的结束而结束

    主进程创建守护进程：

    • 守护进程会在主进程代码执行结束后就终止
    • 守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

    注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止

    (2)守护进程实例

     1 // 守护进程的启动
 2 // 主进程代码执行完后守护进程立即结束
 3 import time
 4 from multiprocessing import Process
 5 
 6 
 7 def func():
 8     while True:
 9         time.sleep(0.5)
10         print("我还活着")
11 
12 
13 def func2():
14     print("in func2 start")
15     time.sleep(8)
16     print("in func2 finish")
17 
18 
19 if __name__ == '__main__':
20     p = Process(target=func)
21     p.daemon = True                     # 设置子进程为守护进程
22     p.start()
23     Process(target=func2).start()       # 开启普通进程
24     i = 0
25     while i < 5:
26         print("我是socket server")
27         time.sleep(1)
28         i += 1
29 
30 # 守护进程 会随着 主进程的代码执行完毕 而结束

    4.进程同步

    (1)进程锁

    多进程实现了程序的异步，让多个任务可以同时在几个进程中并发处理，他们之间的运行没有顺序，一旦开启也不受我们控制。尽管并发编程让我们能更加充分的利用IO资源，但是也给我们带来了新的问题：

    当多个进程使用同一份数据资源的时候，就会引发数据安全或顺序混乱问题

    进程锁实例 - 买票：

    ticket：

    1 {"ticket": 1}

    lock.py：

     1 # 锁 -> 购买火车票
 2 import json
 3 import time
 4 from multiprocessing import Process
 5 from multiprocessing import Lock
 6 
 7 def buy_ticket(i, lock):
 8     lock.acquire()  # 拿钥匙进门
 9     with open('ticket') as f:
10         dic = json.load(f)
11         time.sleep(0.1)
12     if dic['ticket'] > 0:
13         dic['ticket'] -= 1
14         print('33[32m%s买到票了33[0m' % i)
15     else:
16         print('33[31m%s没买到票33[0m' % i)
17     time.sleep(0.1)
18     with open('ticket', 'w') as f:
19         json.dump(dic, f)
20     lock.release()  # 还钥匙
21 
22 if __name__ == '__main__':
23     lock = Lock()
24     for i in range(10):
25         p = Process(target=buy_ticket, args=(i, lock))      # 能在某一时刻操作文件的只能有一个进程
26         p.start()

    结果：

    但是注意：

     1 # 加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
 2 虽然可以用文件共享数据实现进程间通信，但问题是：
 3     效率低（共享数据基于文件，而文件是硬盘上的数据）
 4     需要自己加锁处理
 5 
 6 # 因此我们最好找寻一种解决方案能够兼顾：
 7     效率高（多个进程共享一块内存的数据）
 8     帮我们处理好锁问题
 9 # 这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
10 # 队列和管道都是将数据存放于内存中
11 # 队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，我们应该尽量避免使用共享数据12 # 尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。

    (2)信号量

    互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据 。

    假设商场里有4个迷你唱吧，所以同时可以进去4个人，如果来了第五个人就要在外面等待，等到有人出来才能再进去玩。

    实现：

    信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。这是Dijkstra信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。

    信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念

    信号量实例：

     1 import time
 2 import random
 3 from multiprocessing import Process
 4 from multiprocessing import Semaphore
 5 
 6 def ktv(i, sem):
 7     sem.acquire()       # 获取钥匙 -> 类似P操作 -> 申请一个资源 等待信号量
 8     print('%s走进ktv' % i)
 9     time.sleep(random.randint(1, 5))
10     print('%s走出ktv' % i)
11     sem.release()       # 释放钥匙 -> 类似V操作 -> 传送一个资源 释放信号量
12 
13 if __name__ == '__main__':
14     sem = Semaphore(4)              # 最多容纳4个
15     for i in range(20):
16         p = Process(target=ktv, args=(i, sem))
17         p.start()

    总结：信号量是用锁的原理实现的,内置了一个计数器，在同一时间 只能有指定数量的进程执行某一段被控制住的代码

    (3)事件

    python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear

    事件处理的机制：全局定义一个“Flag”，然后通过以下三个方法进行事件处理：

    • clear：将“Flag”设置为False
    • wait：如果“Flag”值为 False那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True那么event.wait 方法时便不再阻塞。
    • set：将“Flag”设置为True
     1 from multiprocessing import Event
 2 
 3 # 通过一个信号 来控制 多个进程 同时 执行或者阻塞   ->  事件
 4 # 一个信号可以使所有的进程都进入阻塞状态   也可以控制所有的进程解除阻塞
 5 e = Event()  # 创建了一个事件
 6 print(e.is_set())   # 查看一个事件的状态, 事件创建后默认被设置成阻塞
 7 e.set()      # 将这个事件的状态改为True
 8 print(e.is_set())
 9 e.wait()     # 是依据e.is_set()的值来决定是否阻塞的
10 print(123456)
11 e.clear()    # 将这个事件的状态改为False
12 print(e.is_set())
13 e.wait()     # 等待 事件的信号被变成True
14 print('*'*10)
15 
16 # 执行结果:
17 # False
18 # True
19 # 123456
20 # False

    事件处理实例：

     1 # 红绿灯事件
 2 import time
 3 import random
 4 from multiprocessing import Event, Process
 5 
 6 def cars(e, i):
 7     if not e.is_set():
 8         print('car%i在等待' % i)
 9         e.wait()  # 阻塞 直到得到一个 事件状态变成 True 的信号
10     print('33[0;32;40mcar%i通过33[0m' % i)
11 
12 def light(e):
13     while True:
14         if e.is_set():
15             e.clear()
16             print('33[31m红灯亮了33[0m')
17         else:
18             e.set()
19             print('33[32m绿灯亮了33[0m')
20         time.sleep(2)
21 
22 if __name__ == '__main__':
23     e = Event()
24     traffic = Process(target=light, args=(e,))
25     traffic.start()
26     for i in range(20):
27         car = Process(target=cars, args=(e, i))
28         car.start()
29         time.sleep(random.random())

    5.进程间通信 - 队列和管道

    进程间通信 - IPC(Inter-Process Communication)

    (1)队列

    创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递

     1 # 队列方法:
 2 Queue([maxsize]) 
 3 创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现
 4 
 5 Queue的实例q具有以下方法：
 6 q.get( [ block [ ,timeout ] ] ) 
 7 返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止
 8 block: 控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）
 9 timeout: 可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常
10 
11 q.get_nowait( )   同q.get(False)方法
12 
13 q.put(item [, block [,timeout ] ] ) 
14 将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止
15 block: 控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）
16 timeout: 指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常
17 
18 q.qsize() 
19 返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。
20 
21 q.empty() 
22 如果调用此方法时 q为空，返回True
23 如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目
24 
25 q.full() 
26 如果q已满，返回为True
27 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）
28 
29 
30 # 其他方法:
31 q.close() 
32 关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将自动调用此方法。
33 关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。
34 
35 q.cancel_join_thread() 
36 不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。
37 
38 q.join_thread() 
39 连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。

    实例 - 父进程和子进程之间的通信：

     1 # 父进程和子进程之间的通信
 2 from multiprocessing import Process, Queue
 3 
 4 
 5 def produce(que):
 6     que.put('hello')
 7 
 8 
 9 def consumer(que):
10     print(que.get())
11 
12 
13 if __name__ == '__main__':
14     q = Queue()
15     p = Process(target=produce, args=(q, ))
16     p.start()
17     c = Process(target=consumer, args=(q, ))
18     c.start()

    实例 - 批量生产数据放入队列再批量获取结果：

     1 # 批量生产数据放入队列再批量获取结果
 2 import os
 3 import time
 4 import multiprocessing
 5 
 6 # 向queue中输入数据的函数
 7 def inputQ(queue):
 8     info = str(os.getpid()) + '(put):' + str(time.asctime())
 9     queue.put(info)
10 
11 # 向queue中输出数据的函数
12 def outputQ(queue):
13     info = queue.get()
14     print('%s%s33[32m%s33[0m' % (str(os.getpid()), '(get):', info))
15 
16 if __name__ == '__main__':
17     multiprocessing.freeze_support()
18     record1 = []        # store input processes
19     record2 = []        # store output processes
20     queue = multiprocessing.Queue(3)
21 
22     # 输入进程
23     for i in range(10):
24         process = multiprocessing.Process(target=inputQ, args=(queue,))
25         process.start()
26         record1.append(process)
27 
28     # 输出进程
29     for i in range(10):
30         process = multiprocessing.Process(target=outputQ, args=(queue,))
31         process.start()
32         record2.append(process)
33 
34     for p in record1:
35         p.join()
36 
37     for p in record2:
38         p.join()

    (2)生产者消费者模型

    生产者消费者模型见此：https://www.cnblogs.com/wyb666/p/9739608.html

    (3)管道

     1 #创建管道的类：
 2 Pipe([duplex]):在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道
 3 
 4 #参数介绍：
 5 dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送
 6 
 7 
 8 #主要方法：
 9 conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
10 conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError
11 
12 
13 #其他方法：
14 conn1.close():关闭连接。如果conn1被垃圾回收，将自动调用此方法
15 conn1.fileno():返回连接使用的整数文件描述符
16 conn1.poll([timeout]):如果连接上的数据可用，返回True
17 　　timeout指定等待的最长时限。如果省略此参数，方法将立即返回结果。如果将timeout射成None，操作将无限期地等待数据到达
18 
19 conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息
20 　　maxlength指定要接收的最大字节数。如果进入的消息，超过了这个最大值，将引发IOError异常，并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭，再也不存在任何数据，将引发EOFError异常
21 
22 conn.send_bytes(buffer [, offset [, size]])：通过连接发送字节数据缓冲区
23 　　buffer是支持缓冲区接口的任意对象
24 　　offset是缓冲区中的字节偏移量
25 　　size是要发送字节数。结果数据以单条消息的形式发出，然后调用c.recv_bytes()函数进行接收 
26 
27 conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息，并把它保存在buffer对象中，该对象支持可写入的缓冲区接口（即bytearray对象或类似的对象）
28 　　offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间，将引发BufferTooShort异常
     1 # 管道简单使用
 2 from multiprocessing import Pipe, Process
 3 
 4 def func(conn):
 5     conn.send('吃了么')
 6 
 7 if __name__ == '__main__':
 8     conn1, conn2 = Pipe()
 9     Process(target=func, args=(conn1, )).start()
10     print(conn2.recv())
     1 # EOFError
 2 from multiprocessing import Pipe, Process
 3 
 4 def func(conn1, conn2):
 5     conn2.close()
 6     while True:
 7         try:
 8             msg = conn1.recv()
 9             print(msg)
10         except EOFError:
11             conn1.close()
12             break
13 
14 if __name__ == '__main__':
15     conn1, conn2 = Pipe()
16     Process(target=func, args=(conn1, conn2)).start()
17     conn1.close()
18     for i in range(20):
19         conn2.send('吃了么')
20     conn2.close()

    6.进程间数据共享 - Manager

    (1)进程间通信

    展望未来，基于消息传递的并发编程是大势所趋

    即便是使用线程，推荐做法也是将程序设计为大量独立的线程集合，通过消息队列交换数据。

    这样极大地减少了对使用锁定和其他同步手段的需求，还可以扩展到分布式系统中。

    进程间应尽量避免通信，即便需要通信也应该选择进程安全的工具(比如队列)来避免加锁带来的问题。当然可以使用数据库来解决现在进程之间的数据共享问题。

    (2)Manager模块

    进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的；虽然进程间数据独立，但可以通过Manager实现数据共享，事实上Manager的功能远不止于此

     1 # Manager实例:
 2 from multiprocessing import Manager, Process, Lock
 3 
 4 def main(dic, lock):        # 加锁确保进程之间的数据共享万无一失
 5     lock.acquire()
 6     dic['count'] -= 1
 7     lock.release()
 8 
 9 if __name__ == '__main__':
10     m = Manager()
11     l = Lock()
12     dic = m.dict({'count': 100})
13     p_lst = []
14     for i in range(50):
15         p = Process(target=main, args=(dic, l))
16         p.start()
17         p_lst.append(p)
18     for i in p_lst:
19         i.join()
20     print('主进程', dic)

    当然上面的写法也可以更高级：

     1 # Manager实例
 2 from multiprocessing import Manager, Process, Lock
 3 def work(d, lock):
 4     with lock:  # 不加锁而操作共享的数据,肯定会出现数据错乱
 5         d['count'] -= 1
 6 
 7 if __name__ == '__main__':
 8     lock=Lock()
 9     with Manager() as m:
10         dic=m.dict({'count':100})
11         p_l=[]
12         for i in range(100):
13             p=Process(target=work,args=(dic,lock))
14             p_l.append(p)
15             p.start()
16         for p in p_l:
17             p.join()
18         print(dic)

    7.进程池 

    (1)进程池的概念

    在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？

    • 首先创建进程需要消耗时间，销毁进程也需要消耗时间
    • 另外即便开启了成千上万的进程，操作系统也不能让他们同时执行，这样反而会影响程序的效率。因此不能无限制的根据任务开启或者结束进程

    于是就有了进程池的概念：

    • 定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务
    • 如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行
    • 也就是说池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果

    (2)multiprocess.Pool模块

     1 Pool([numprocess  [,initializer [, initargs]]]):创建进程池
 2 　　numprocess: 要创建的进程数，如果省略，将默认使用cpu_count()的值
 3 　　initializer: 是每个工作进程启动时要执行的可调用对象，默认为None
 4 　　initargs: 是要传给initializer的参数组
 5 　　
 6 
 7 基本方法:
 8 　　p. map(func, iterable, chunksize=None):  第一个参数是函数，第二个参数是一个迭代器，将迭代器中的数字作为参数依次传入函数中执行，返回值为所有结果的[]
 9 　　p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果
10 　　'''需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()'''
11 
12 　　p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果
13 　　'''此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。'''
14    
15 　　p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成 
16 　　P.join():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用
17 
18 
19 其他方法:
20 　　方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。
21 　　实例obj具有以下方法:
22 　　　　obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。
23 　　　　obj.ready():如果调用完成，返回True
24 　　　　obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常
25 　　　　obj.wait([timeout]):等待结果变为可用。
26 　　　　obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数

    (3)代码实例

    进程池和多进程效率对比：

     1 import time
 2 from multiprocessing import Pool, Process
 3 
 4 def func(n):
 5     for v in range(10):
 6         n+1
 7 
 8 if __name__ == '__main__':
 9     start = time.time()
10     pool = Pool(5)
11     pool.map(func, range(100))
12     t1 = time.time() - start
13 
14     start = time.time()
15     p_list = []
16     for i in range(100):
17         p = Process(target=func, args=(i, ))
18         p_list.append(p)
19         p.start()
20     for p in p_list:
21         p.join()
22     t2 = time.time() - start
23     print(t1, t2)

    结果如下：

    进程池的同步调用和异步调用：

     1 import os
 2 import time
 3 from multiprocessing import Pool
 4 
 5 def func(n):
 6     print("start func %s" % n, os.getpid())
 7     time.sleep(1)
 8     print("end func %s" % n, os.getpid())
 9     return n*2
10 
11 # 测试进程池的apply方法 -> 同步调用
12 def test_apply(p):
13     res_list = []
14     for i in range(10):
15         res = p.apply(func, args=(i, ))
16         res_list.append(res)
17     print(res_list)
18 
19 # 测试进程池的apply_async方法 -> 异步调用
20 def test_apply_async(p):
21     res_list = []
22     for i in range(10):
23         res = p.apply_async(func, args=(i, ))
24         res_list.append(res)
25         # 异步运行，根据进程池中有的进程数，每次最多5个子进程在异步执行  返回结果之后，将结果放入列表，归还进程，之后再执行新的任务
26         # 需要注意的是，进程池中的三个进程不会同时开启或者同时结束  而是执行完一个就释放一个进程，这个进程就去接收新的任务 
27         # 异步apply_async用法：如果使用异步提交的任务，主进程需要使用join，等待进程池内任务都处理完，然后可以用get收集结果
28         # 否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了
29     p.close()
30     p.join()
31     for res in res_list:
32         print(res, res.get())
33         # 使用get来获取apply_async的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
34 
35 if __name__ == '__main__':
36     pool = Pool(5)
37     test_apply(pool)
38     # test_apply_async(pool)

    进程池的回调函数：

    1 # 关于回调函数
2 # 需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数
3 
4 # 我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果
     1 # 下面的回调函数 -> 先执行func1 然后把func1的结果作为参数传给func2
 2 import os
 3 from multiprocessing import Pool
 4 
 5 def func1(n):                           # func1在子进程中执行
 6     print('in func1', os.getpid())
 7     return n * n
 8 
 9 def func2(nn):                          # func2在主进程中执行
10     print('in func2', os.getpid())
11     print(nn)
12 
13 if __name__ == '__main__':
14     print('主进程 :', os.getpid())
15     p = Pool(5)
16     for i in range(10):
17         p.apply_async(func1, args=(10,), callback=func2)
18     p.close()
19     p.join()
  • 相关阅读:
    生物神经网络和人工神经网络浅谈
    卷积神经网络
    DOM进阶之HTMl属性操作（对象属性）
    01 selenium基本使用补充
    01 selenium基本使用
    day4笔记
    03 获取豆瓣电影top250
    02 爬取视频
    day3笔记
    01 requests基本使用
  • 原文地址：https://www.cnblogs.com/wyb666/p/9065215.html
Copyright © 2011-2022 走看看