python多进程并发之multiprocessing

时间：2022-02-07 03:33:18

multiprocessing.Process

multiprocessing包是Python中的多进程管理包。它与 threading.Thread类似，可以利用multiprocessing.Process对象来创建一个进程。该进程可以允许放在Python程序内部编写的函数中。该Process对象与Thread对象的用法相同，拥有is_alive()、join([timeout])、run()、start()、terminate()等方法。属性有：authkey、daemon（要通过start()设置）、exitcode(进程在运行时为None、如果为–N，表示被信号N结束）、name、pid。此外multiprocessing包中也有Lock/Event/Semaphore/Condition类，用来同步进程，其用法也与threading包中的同名类一样。multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

这个模块表示像线程一样管理进程，这个是multiprocessing的核心，它与threading很相似，对多核CPU的利用率会比threading好的多。

看一下Process类的构造方法：

__init__(self, group=None, target=None, name=None, args=(), kwargs={})

参数说明：

  • group：进程所属组。基本不用
  • target：表示调用对象。
  • args：表示调用对象的位置参数元组。
  • name：别名
  • kwargs：表示调用对象的字典。

创建进程的简单实例：

#coding=utf-8
import multiprocessing

def do(n) :
#获取当前线程的名字
  name = multiprocessing.current_process().name
print name,'starting'
print "worker ", n
return 

if __name__ == '__main__' :
  numList = []
for i in xrange(5) :
    p = multiprocessing.Process(target=do, args=(i,))
    numList.append(p)
    p.start()
    p.join()
print "Process end."

执行结果：

Process-1 starting
worker  0
Process end.
Process-2 starting
worker  1
Process end.
Process-3 starting
worker  2
Process end.
Process-4 starting
worker  3
Process end.
Process-5 starting
worker  4
Process end.

创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，并用其start()方法启动，join()方法表示等待子进程结束以后再继续往下运行，通常用于进程间的同步。

注意：
在Windows上要想使用进程模块，就必须把有关进程的代码写在当前.py文件的if __name__ == ‘__main__’ :语句的下面，才能正常使用Windows下的进程模块。Unix/Linux下则不需要。


multiprocess.Pool

当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个目标，手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。

Pool可以提供指定数量的进程供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来它。

apply_async和apply

函数原型：

apply_async(func[, args=()[, kwds={}[, callback=None]]])

二者都是向进程池中添加新的进程，不同的时，apply每次添加新的进程时，主进程和新的进程会并行执行，但是主进程会阻塞，直到新进程的函数执行结束。 这是很低效的，所以python3.x之后不再使用

apply_async和apply功能相同，但是主进程不会阻塞。

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

import multiprocessing
import time

def func(msg):
print "*msg: ", msg
    time.sleep(3)
print "*end"

if __name__ == "__main__":
# 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
    pool = multiprocessing.Pool(processes=3)
for i in range(10):
        msg = "hello [{}]".format(i)
# pool.apply(func, (msg,))
        pool.apply_async(func, (msg,))        # 异步开启进程, 非阻塞型, 能够向池中添加进程而不等待其执行完毕就能再次执行循环

print "--" * 10
    pool.close()   # 关闭pool, 则不会有新的进程添加进去
    pool.join()    # 必须在join之前close, 然后join等待pool中所有的线程执行完毕
print "All process done."

运行结果：

"D:\Program Files\Anaconda2\python.exe" E:/pycharm/test/multiprocessing/v1.py
--------------------
*msg:  hello [0]
*msg:  hello [1]
*msg:  hello [2]
*end
*msg:  hello [3]
*end
*end
*msg:  hello [4]
*msg:  hello [5]
*end
*msg:  hello [6]
*end
*end
*msg:  hello [7]
*msg:  hello [8]
*end
*msg:  hello [9]
*end*end

*end
All process done.

Process finished with exit code 0


获得进程的执行结果

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

import multiprocessing
import time

def func_with_return(msg):
print "*msg: ", msg
    time.sleep(3)
print "*end"
return "{} return".format(msg)

if __name__ == "__main__":
# 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
    pool = multiprocessing.Pool(processes=3)
    results = []
for i in range(10):
        msg = "hello [{}]".format(i)
        res = pool.apply_async(func_with_return, (msg,))        # 异步开启进程, 非阻塞型, 能够向池中添加进程而不等待其执行完毕就能再次执行循环
        results.append(res)

print "--" * 10
    pool.close()   # 关闭pool, 则不会有新的进程添加进去
    pool.join()    # 必须在join之前close, 然后join等待pool中所有的线程执行完毕
print "All process done."

print "Return results: "
for i in results:
print i.get()   # 获得进程的执行结果

结果：

"D:\Program Files\Anaconda2\python.exe" E:/pycharm/test/multiprocessing/v1.py
--------------------
*msg:  hello [0]
*msg:  hello [1]
*msg:  hello [2]
*end
*end
*msg:  hello [3]
*msg:  hello [4]
*end
*msg:  hello [5]
*end
*end
*msg:  hello [6]
*msg:  hello [7]
*end
*msg:  hello [8]
*end
*end
*msg:  hello [9]
*end
*end
All process done.
Return results: 
hello [0] return
hello [1] return
hello [2] return
hello [3] return
hello [4] return
hello [5] return
hello [6] return
hello [7] return
hello [8] return
hello [9] return

Process finished with exit code 0


map

函数原型：

map(func, iterable[, chunksize=None])

Pool类中的map方法，与内置的map函数用法行为基本一致，它会使进程阻塞直到返回结果。
注意，虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程。

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

import multiprocessing
import time

def func_with_return(msg):
print "*msg: ", msg
    time.sleep(3)
print "*end"
return "{} return".format(msg)

if __name__ == "__main__":
# 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
    pool = multiprocessing.Pool(processes=3)
    results = []
    msgs = []
for i in range(10):
        msg = "hello [{}]".format(i)
        msgs.append(msg)

    results = pool.map(func_with_return, msgs)

print "--" * 10
    pool.close()   # 关闭pool, 则不会有新的进程添加进去
    pool.join()    # 必须在join之前close, 然后join等待pool中所有的线程执行完毕
print "All process done."

print "Return results: "
for i in results:
print i   # 获得进程的执行结果

执行结果：

"D:\Program Files\Anaconda2\python.exe" E:/pycharm/test/multiprocessing/v2.py
*msg:  hello [0]
*msg:  hello [1]
*msg:  hello [2]
*end*end

*msg:  hello [3]
*msg:  hello [4]
*end
*msg:  hello [5]
*end*end

*msg:  hello [6]
*msg:  hello [7]
*end
*msg:  hello [8]
*end
*end
*msg:  hello [9]
*end
*end
--------------------
All process done.
Return results: 
hello [0] return
hello [1] return
hello [2] return
hello [3] return
hello [4] return
hello [5] return
hello [6] return
hello [7] return
hello [8] return
hello [9] return

Process finished with exit code 0

注意执行结果中“—-”的位置，可以看到，map之后，主进程是阻塞的，等待map的结果返回

close()

关闭进程池（pool），使其不在接受新的任务。

terminate()

结束工作进程，不在处理未处理的任务。

join()

主进程阻塞等待子进程的退出，join方法必须在close或terminate之后使用。


进程间通信

多进程最麻烦的地方就是进程间通信，IPC比线程通信要难处理的多，所以留作单独一篇来记录
标签：

相关文章