python3之multiprocess模块(上)

基础方法

简单的使用方法：

  1: import multiprocessing
  2: def worker():
  3:     print("working!")
  4: if __name__ == '__main__':
  5:     jobs = []
  6:     for i in range(5):
  7:         p = multiprocessing.Process(target=worker)
  8:         jobs.append(p)
  9:         p.start()

另外一种使用方法是导入第三方文件的目标函数：

work.py

  1: def worker():
  2:     print("worker")
  3:     return

multiprocess_main.py

  1: import multiprocessing
  2: import worker
  3: if __name__ == '__main__':
  4:     jobs = []
  5:     for i in range(5):
  6:         p = multiprocessing.Process(
  7:             target=worker.worker,
  8:         )
  9:         jobs.append(p)
 10:         p.start()

要传递参数的话，这个参数必须能pickle序列化。

  1: import multiprocessing
  2: def worker(i):
  3:     print("worker%s" % i)
  4: if __name__ == '__main__':
  5:     jobs = []
  6:     for i in range(5):
  7:         p = multiprocessing.Process(target=worker, args=(i,))
  8:         jobs.append(p)
  9:         p.start()

打印结果：

  1: worker0
  2: worker1
  3: worker2
  4: worker3
  5: worker4

确定当前的进程

默认的name

  1: name = multiprocessing.current_process().name

或者实例化时传入name参数

  1: t = multiprocessing.Process(
  2:     name="t1",
  3:     target=work,
  4: )

守护进程

类似于守护线程，守护进程会在主程序退出之前自动终止，以免留下孤儿进程。

t在这里是某个进程对象。

  1: t.daemon = True

同样的要等待一个进程完成，可以使用join()方法

  1: t.join()

默认地join()会无限阻塞。可以传入一个超时参数(浮点数，秒数)。在这个时限内即使没有完成，join()也会停止阻塞，返回

  1: t.join(sec)

终止进程

对一个进程对象调用terminate()会结束子进程。

  1: t.terminate()

进程退出时生成的状态码可以通过exitcode属性访问。下表是退出码：

  1: ==0 未生成任何错误
  2: > 0 进程有一个错误，并以该错误码退出
  3: < 0 进程以一个-1*exitcode信号结束

exitcode是属性不是方法

  1: t.exitcode

日志

使用multiprocess.log_to_stderr()属性基于logging模块来建一个日志记录的对象，里面的参数传入日志的级别。要注意设置的位置。

  1: if __name=='__main__':
  2:     multiprocess.log_to_stderr(logging.DEBUG)
  3:     p = multiprocess.Process(target=worker)
  4:     p.start()
  5:     p.join()

也可以直接拿到这个日志对象来处理。

  1: logger = multiprocess.get_logger()
  2: logger.setLevel(logging.INFO)

派生进程

最简单的方法是使用Process并传入一个目标函数，但也可以使用一个定制子类。但子类应该覆盖run()方法以完成工作

  1: import multiprocessing
  2: class Worker(multiprocessing.Process):
  3:     def run(self):
  4:         print("In{}".format(self.name))
  5:         return
  6: if __name__ == '__main__':
  7:     jobs = []
  8:     for i in range(5):
  9:         p = Worker()
 10:         jobs.append(p)
 11:         p.start()
 12:     for j in jobs:
 13:         j.join()

进程间的通信

用一个multiprocessing.Queue([maxsize])来回传递消息，能够pickle串行化的任何对象都可以通过Queue传递。

q.put方法用以插入数据到队列中。

q.get方法可以从队列读取并且删除一个元素。

简单的使用方法：

  1: import multiprocessing
  2: q = multiprocessing.Queue(3)
  3: 
  4: q.put(1)
  5: q.put(2)
  6: q.put(3)
  7: # 满了,返回True
  8: print(q.full())
  9: 
 10: print(q.get())
 11: print(q.get())
 12: print(q.get())
 13: # 空了，返回True
 14: print(q.empty())

生产者-消费者模型

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

  1: import multiprocessing
  2: import time,random,os
  3: def consumer(q, name):
  4:     while True:
  5:         res = q.get()
  6:         time.sleep(random.randint(1, 3))
  7:         print("%s 吃了 %s" % (name, res))
  8: def producer(q,name,food):
  9:     for i in range(3):
 10:         time.sleep(random.randint(1, 3))
 11:         res = "%s%s" % (food, i)
 12:         q.put(res)
 13:         print("%s 生产了 %s" % (name, res))
 14: if __name__ == '__main__':
 15:     q = multiprocessing.Queue()
 16:     # 生产者
 17:     p1 = multiprocessing.Process(target=producer, args=(q, "egon", "包子"))
 18:     # 消费者
 19:     c1 = multiprocessing.Process(target=consumer, args=(q, "alex"))
 20:     p1.start()
 21:     c1.start()
 22:     print("主")

但是上述代码在三个包子被消费完后会卡死，所以需要发送结束信号，而JoinableQueue这种队列提供了这种机制。

JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：

q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常 q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

  1: from multiprocessing import Process,JoinableQueue
  2: import time,random
  3: def consumer(q,name):
  4:     while True:
  5:         res=q.get()
  6:         time.sleep(random.randint(1,3))
  7:         print('%s 吃 %s' %(name,res))
  8:         q.task_done() #发送信号给q.join()，说明已经从队列中取走一个数据并处理完毕了
  9: def producer(q,name,food):
 10:     for i in range(3):
 11:         time.sleep(random.randint(1,3))
 12:         res='%s%s' %(food,i)
 13:         q.put(res)
 14:         print('%s 生产了 %s' %(name,res))
 15:     q.join() #等到消费者把自己放入队列中的所有的数据都取走之后，生产者才结束
 16: if __name__ == '__main__':
 17:     q=JoinableQueue() #使用JoinableQueue()
 18: 
 19:     #生产者们
 20:     p1=Process(target=producer,args=(q,'egon1','包子'))
 21:     p2=Process(target=producer,args=(q,'egon2','骨头'))
 22:     p3=Process(target=producer,args=(q,'egon3','泔水'))
 23: 
 24:     #消费者们
 25:     c1=Process(target=consumer,args=(q,'alex1'))
 26:     c2=Process(target=consumer,args=(q,'alex2'))
 27:     c1.daemon=True
 28:     c2.daemon=True
 29: 
 30:     #开始
 31:     p1.start()
 32:     p2.start()
 33:     p3.start()
 34:     c1.start()
 35:     c2.start()
 36: 
 37:     p1.join()
 38:     p2.join()
 39:     p3.join()
 40:     #1、主进程等生产者p1、p2、p3结束
 41:     #2、而p1、p2、p3是在消费者把所有数据都取干净之后才会结束
 42:     #3、所以一旦p1、p2、p3结束了，证明消费者也没必要存在了，应该随着主进程一块死掉，因而需要将消费者们设置成守护进程
 43:     print('主process')

结果：

  1: egon2 生产了 骨头0
  2: egon1 生产了 包子0
  3: egon3 生产了 泔水0
  4: alex1 吃 骨头0
  5: egon3 生产了 泔水1
  6: egon2 生产了 骨头1
  7: alex2 吃 包子0
  8: egon1 生产了 包子1
  9: alex1 吃 泔水0
 10: egon2 生产了 骨头2
 11: egon1 生产了 包子2
 12: egon3 生产了 泔水2
 13: alex1 吃 骨头1
 14: alex2 吃 泔水1
 15: alex2 吃 骨头2
 16: alex1 吃 包子1
 17: alex1 吃 泔水2
 18: alex2 吃 包子2
 19: 主process