目录

• 1、apply_async
• 2、Manager
• 3、Pool进程池
• 4、queue资源共享
• 5、Pipe
• 6、joinableQueue
• 7、Value_Array进程资源共享

1、apply_async

# encoding:utf-8 # -*- coding:utf-8 -*- # 异步进程池（非阻塞） from multiprocessing import Pool def test(i): print(i) if __name__ == "__main__": pool = Pool(8) for i in range(100): pool.apply_async(test, args=(i,)) # 结果是乱序的 #子进程和父进程是异步的 print("test") pool.close() pool.join() 

2、Manager

# encoding:utf-8 import multiprocessing # Manager()返回的manager对象控制了一个server进程， # 此进程包含的python对象可以被其他的进程通过proxies来访问。 # 从而达到多进程间数据通信且安全。Manager模块常与Pool模块一起使用。 # 以下类型均不是进程安全的，需要加锁. # Array(self,*args,kwds) # BoundedSemaphore(self,*args,kwds) # Condition(self,*args,kwds) # Event(self,*args,kwds) # JoinableQueue(self,*args,kwds) # Lock(self,*args,kwds) # Namespace(self,*args,kwds) # Pool(self,*args,kwds) # Queue(self,*args,kwds) # RLock(self,*args,kwds) # Semaphore(self,*args,kwds) # Value(self,*args,kwds) # dict(self,*args,kwds) # list(self,*args,kwds) def f(x, arr, l, d, n): x.value = 3.14 arr[0] = 5 l.append('xx') d[1] = 2 n.a = 10 if __name__ == '__main__': server = multiprocessing.Manager() x = server.Value('d', 0.0) arr = server.Array('i', range(10)) l = server.list() d =server.dict() n = server.Namespace() proc = multiprocessing.Process(target=f, args=(x, arr, l, d, n)) proc.start() proc.join() print(x.value) print(arr) print(l) print(d) print(n) 

3、Pool进程池

# encoding:utf-8 # -*- coding:utf-8 -*- # Pool+map import datetime import os from multiprocessing import Pool from time import sleep def test(i): print(i) sleep(5) print('Run child process (%s)...' % ( os.getpid())) if __name__ == "__main__": lists = range(30) pool = Pool(8) # 控制每时只有8个进程 t1=datetime.datetime.now() pool.map(test, lists) pool.close() # close() 关闭pool，使其不在接受新的任务 pool.join() # 主进程阻塞，等待子进程的退出 t2 = datetime.datetime.now() print(f'cost==={ 
     t2-t1}') 

4、queue资源共享

 # 普通的全局变量是不能被子进程所共享的，只有通过Multiprocessing组件构造的数据结构可以被共享。 # Queue是用来创建进程间资源共享的队列的类，使用Queue可以达到多进程间数据传递的功能（缺点：只适用Process类，不能在Pool进程池中使用）。 from multiprocessing import Process, Queue import os import time import random def write(q): print("process to write:{}" .format(os.getpid())) for value in [1, 2, 3]: print("put {} to queue..." .format(value)) q.put(value) time.sleep(random.random()) def read(q): print("process to read:{}" .format(os.getpid())) while True: value = q.get(True) print("get {} from queue..." .format(value)) if __name__ == '__main__': q = Queue() p_w = Process(target=write, args=(q,)) p_r = Process(target=read, args=(q,)) p_w.start() p_r.start() p_w.join() # 等待pw结束 p_r.terminate() # pr进程里是死循环，无法等待其结束，只能强行终止 

5、Pipe

# encoding:utf-8 from multiprocessing import Process, Pipe import time # Pipe可以在进程之间创建一条管道，并返回组（conn1,conn2）, # 其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道。 # 子进程执行方法 def f(Subconn): time.sleep(1) Subconn.send("吃了吗") print("来自父亲的问候:{}" .format(Subconn.recv())) Subconn.close() if __name__ == '__main__': parent_conn, child_conn = Pipe() # 创建管道两端 p = Process(target=f, args=(child_conn,)) # 创建子进程 p.start() print("来自儿子的问候：{}" .format(parent_conn.recv())) parent_conn.send("嗯") 

6、joinableQueue

# encoding:utf-8 from multiprocessing import Process, JoinableQueue # JoinableQueue就像是一个Queue对象， # 但队列允许项目的使用者通知生成者项目已经被成功处理。 # 通知进程是使用共享的信号和条件变量来实现的。 import time, random # 消费者 接收到生产者的join请求后执行task_done def consumer(q): while True: res = q.get() print('消费者拿到了 %s' % res) q.task_done() # 生产者 生产完了负责发送join  def producer(seq, q): for item in seq: time.sleep(random.randrange(1,2)) q.put(item) print('生产者做好了 %s' % item) q.join() if __name__ == "__main__": q = JoinableQueue() seq = ('产品%s' % i for i in range(5)) p = Process(target=consumer, args=(q,)) p.daemon = True # 设置为守护进程，在主线程停止时p也停止，但是不用担心，producer内调用q.join保证了consumer已经处理完队列中的所有素 p.start() producer(seq, q) print('主线程') 

7、Value_Array进程资源共享

# encoding:utf-8 # 用于进程通信，资源共享，注意：Value和Array只适用于Process类。 # multiprocessing 中Value和Array的实现原理都是在共享内存中创建ctypes()对象来达到共享数据的目的，两者实现方法大同小异，只是选用不同的ctypes数据类型而已。 import multiprocessing def f(n, a): n.value = 3.14 a[0] = 5 if __name__ == '__main__': num = multiprocessing.Value('d', 0.0) arr = multiprocessing.Array('i', range(10)) p = multiprocessing.Process(target=f, args=(num, arr)) p.start() p.join() print(num.value) print(arr[:])