AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

【Python】浅谈 multiprocessing

2017-07-02 1161
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 一前言     使用python进行并发处理多台机器/多个实例的时候，我们可以使用threading ，但是由于著名的GIL存在，实际上threading 并未提供真正有效的并发处理，要充分利用到多核CPU，我们需要使用多进程。
一前言 
   使用python进行并发处理多台机器/多个实例的时候，我们可以使用threading ，但是由于著名的 GIL存在，实际上threading 并未提供真正有效的并发处理，要充分利用到多核CPU，我们需要使用多进程。Python提供了非常好用的多进程包-- multiprocessing。multiprocessing 可以利用multiprocessing.Process对象来创建一个进程,该Process对象与Threading对象的用法基本相同，具有相同的方法(官方原话:" The multiprocessing package mostly replicates the API of the threading module.") 比如:start(),run(),join()的方法。multiprocessing包中也有Lock/Event/Semaphore/Condition/Pipe/Queue类用于进程之间的通信。话不多说 show me the code！

二使用
2.1 初识异同
下面的程序显示threading和multiprocessing的在使用方面的异同，相近的函数join()，start(),append() 等，并做同一件事情打印自己的进程pid

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. import os
  4. import threading
  5. import multiprocessing
  6. def printer(msg):
  7.     print(msg, os.getpid())
  8. print('Main begin:', os.getpid())
  9. # threading
  10. record = []
  11. for i in range(5):
  12.     thread = threading.Thread(target=printer, args=('threading',))
  13.     thread.start()
  14.     record.append(thread)
  15. for thread in record:
  16.     thread.join()
  17. # multi-process
  18. record = []
  19. for i in range(5):
  20.     process = multiprocessing.Process(target=printer, args=('multiprocessing',))
  21.     process.start()
  22.     record.append(process)
  23. for process in record:
  24.     process.join()
  25. print('Main end:', os.getpid())
输出结果

点击(此处)折叠或打开

  1. Main begin: 9524
  2. threading 9524
  3. threading 9524
  4. threading 9524
  5. threading 9524
  6. threading 9524
  7. multiprocessing 9539
  8. multiprocessing 9540
  9. multiprocessing 9541
  10. multiprocessing 9542
  11. multiprocessing 9543
  12. Main end: 9524
从例子的结果可以看出多线程threading的进程id和主进程(父进程)pid一样 ，同为9524; 多进程打印的pid每个都不一样,for循环中每创建一个process对象都年开一个进程。其他相关的方法基本类似。

2.2 用法
创建进程的类:
Process([group [, target [, name [, args [, kwargs]]]]])，
target表示调用对象，
args表示调用对象的位置参数元组。
kwargs表示调用对象的字典。
name为进程的别名。
group实质上不使用，为None。
方法:is_alive()、join([timeout])、run()、start()、terminate()。其中，Process以start()启动某个进程，并自动调用run方法.
属性:authkey、daemon（要通过start()设置，必须设置在方法start之前）、exitcode(进程在运行时为None、如果为–N，表示被信号N结束）、name、pid。其中daemon是父进程终止后自动终止，且自己不能产生新进程，必须在start()之前设置。

2.3 创建单进程
单线程比较简单,创建一个 Process的实例对象就好，传入参数 target 为已经定义好的方法worker以及worker需要的参数

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. """
  4. author: yangyi@youzan.com
  5. time: 2017/7/2 下午6:45
  6. func:
  7. """
  8. import multiprocessing
  9. import datetime, time
  10. def worker(interval):
  11.     print("process start: {0}".format(datetime.datetime.today()));
  12.     time.sleep(interval)
  13.     print("process end: {0}".format(datetime.datetime.today()));

  15. if __name__ == "__main__":
  16.     p = multiprocessing.Process(target=worker, args=(5,))
  17.     p.start()
  18.     p.join()
  19.     print "end!"
2.4 创建多进程

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. """
  4. author: yangyi@youzan.com
  5. time: 2017/7/2 下午7:50
  6. func:
  7. """
  8. import multiprocessing
  9. def worker(num):
  10.     print "worker %d" %num


  13. if __name__ == "__main__":
  14.     print("The number of CPU is:" + str(multiprocessing.cpu_count()))
  15.     proc = []
  16.     for i in xrange(5):
  17.         p = multiprocessing.Process(target=worker, args=(i,))
  18.         proc.append(p)
  19.     for p in proc:
  20.         p.start()
  21.     for p in proc:
  22.         p.join()
  23.     print "end ..."
输出

点击(此处)折叠或打开

  1. The number of CPU is:4
  2. worker 0
  3. worker 1
  4. worker 2
  5. worker 3
  6. worker 4
  7. main process end ...
2.5 线程池
multiprocessing提供进程池的类--Pool，它可以指定程序最大可以调用的进程数量，当有新的请求提交到pool中时，如果进程池还没有满，那么就会创建一个新的进程用来执行该请求；但如果进程池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来它。
构造方法:
Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])
processes  : 使用的工作进程的数量，如果processes是None,默认使用os.cpu_count()返回的数量。
initializer: 如果initializer是None，那么每一个工作进程在开始的时候会调用initializer(*initargs)。
maxtasksperchild:工作进程退出之前可以完成的任务数，完成后用一个新的工作进程来替代原进程，来让闲置的资源被释放。maxtasksperchild默认是None，意味着只要Pool存在工作进程就会一直存活。
context: 用在制定工作进程启动时的上下文，一般使用multiprocessing.Pool()或者一个context对象的Pool()方法来创建一个池，两种方法都适当的设置了context。

实例方法:
　　apply(func[, args[, kwds]]):同步进程池
　　apply_async(func[, args[, kwds[, callback[, error_callback]]]]) :异步进程池
　　close() : 关闭进程池，阻止更多的任务提交到pool，待任务完成后，工作进程会退出。
　　terminate() : 结束工作进程，不在处理未完成的任务.
　　join() : 等待工作线程的退出,在调用join()前必须调用close()或者 terminate(),因为被终止的进程需要被父进程调用wait（join等价与wait），否则进程会成为僵尸进程。

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. """
  4. author: yangyi@youzan.com
  5. time: 2017/7/2 下午7:50
  6. func:
  7. """
  8. from multiprocessing import Pool
  9. import time
  10. def worker(num):
  11.     print "worker %d" %num
  12.     time.sleep(2)
  13.     print "end worker %d" %num

  15. if __name__ == "__main__":
  16.     proc_pool = Pool(2)
  17.     for i in xrange(4):
  18.         proc_pool.apply_async(worker, (i,)) #使用了异步调用，从输出结果可以看出来

  20.     proc_pool.close()
  21.     proc_pool.join()
  22.     print "main process end ..."
输出结果

点击(此处)折叠或打开

  1. worker 0
  2. worker 1
  3. end worker 0
  4. end worker 1
  5. worker 2
  6. worker 3
  7. end worker 2
  8. end worker 3
  9. main process end ..
解释:创建一个进程池pool 对象proc_pool,并设定进程的数量为2，xrange(4)会相继产生四个对象[0, 1, 2, 4]，四个对象被提交到pool中，因pool指定进程数为2，所以0、1会直接送到进程中执行，当其中的2个任务执行完之后才空出2进程处理对象2和3,所以会出现输出  worker 2 worker 3 出现在end worker 0 end worker 1之后。思考一下如果调用   proc_pool.apply(worker, (i,)) 的输出结果会是什么样的？

2.6 使用queue
multiprocessing提供队列类，可以通过调用multiprocessing.Queue(maxsize) 初始化队列对象，maxsize表示队列里面最多的元素个数。
例子 创建了两个函数入队，出队，出队处理时使用了lock特性，串行化取数据。
  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. """
  4. author: yangyi@youzan.com
  5. time: 2017/7/2 下午9:03
  6. func:
  7. """
  8. import time
  9. from multiprocessing import Process, current_process,Lock,Queue
  10. import datetime
  11. def inputQ(queue):
  12.     time.sleep(1)
  13.     info = "proc_name: " + current_process().name + ' was putted in queue at: ' + str(datetime.datetime.today())
  14.     queue.put(info)
  15. def outputQ(queue,lock):
  16.     info = queue.get()
  17.     lock.acquire()
  18.     print ("proc_name: " + current_process().name + ' gets info :' + info)
  19.     lock.release()
  20. if __name__ == '__main__':
  21.     record1 = [] # store input processes
  22.     record2 = [] # store output processes
  23.     lock = Lock() # To prevent messy print
  24.     queue = Queue(3)
  25.     for i in range(10):
  26.         process = Process(target=inputQ, args=(queue,))
  27.         process.start()
  28.         record1.append(process)
  29.     for i in range(10):
  30.         process = Process(target=outputQ, args=(queue,lock))
  31.         process.start()
  32.         record2.append(process)
  33.     for p in record1:
  34.         p.join()
  35.     queue.close() # No more object will come, close the queue
  36.     for p in record2:
  37.         p.join()
2.7 使用pipe 
Pipe可以是单向(half-duplex)，也可以是双向(duplex)。我们通过mutiprocessing.Pipe(duplex=False)创建单向管道 (默认为双向)。一个进程从PIPE一端输入对象，然后被PIPE另一端的进程接收，单向管道只允许管道一端的进程输入，而双向管道则允许从两端输入。
用法 multiprocessing.Pipe([duplex])
该类返回一组对象实例(conn1, conn2)，分别代表发送和接受消息的两端。

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. """
  4. author: yangyi@youzan.com
  5. time: 2017/7/2 下午8:01
  6. func:
  7. """
  8. from multiprocessing import Process, Pipe
  9. def p1(conn, name):
  10.     conn.send('hello ,{name}'.format(name=name))
  11.     print "p1 receive :", conn.recv()
  12.     conn.close()

  14. def p2(conn, name):
  15.     conn.send('hello ,{name}'.format(name=name))
  16.     print "p2 receive :", conn.recv()
  17.     conn.close()

  19. if __name__ == '__main__':
  20.     parent_conn, child_conn = Pipe()
  21.     proc1 = Process(target=p1, args=(child_conn, "parent_conn"))
  22.     proc2 = Process(target=p2, args=(parent_conn, "child_conn"))
  23.     proc1.start()
  24.     proc2.start()
  25.     proc1.join()
  26.     proc2.join()
输出:

点击(此处)折叠或打开

  1. p1 receive : hello ,child_conn
  2. p2 receive : hello ,parent_conn
该例子中 p1 p2 通过pipe 给彼此相互发送信息,p1 发送"parent_conn" 给 p2 ,p2 发送"child_conn" 给p1.
2.8 daemon程序对比结果
  1. import multiprocessing
  2. import datetime, time
  3. def worker(interval):
  4.     print("process start: {0}".format(datetime.datetime.today()));
  5.     time.sleep(interval)
  6.     print("process end: {0}".format(datetime.datetime.today()));
  7. if __name__ == "__main__":
  8.     p = multiprocessing.Process(target=worker, args=(5,))
  9.     p.start()
  10.     print "end!"
输出:

点击(此处)折叠或打开

  1. end!
  2. process start: 2017-07-02 18:47:30.656244
  3. process end: 2017-07-02 18:47:35.657464

设置 daemon  =  True，程序随着主程序结束而不等待子进程。
  1. import multiprocessing
  2. import datetime, time
  3. def worker(interval):
  4.     print("process start: {0}".format(datetime.datetime.today()));
  5.     time.sleep(interval)
  6.     print("process end: {0}".format(datetime.datetime.today()));
  7. if __name__ == "__main__":
  8.     p = multiprocessing.Process(target=worker, args=(5,))
  9.     p.daemon = True
  10.     p.start()
  11.     print "end!"
输出:
end!
因为子进程设置了daemon属性，主进程结束，multiprocessing创建的进程对象就随着结束了。

  1. import multiprocessing
  2. import datetime, time
  3. def worker(interval):
  4.     print("process start: {0}".format(datetime.datetime.today()));
  5.     time.sleep(interval)
  6.     print("process end: {0}".format(datetime.datetime.today()));
  7. if __name__ == "__main__":
  8.     p = multiprocessing.Process(target=worker, args=(5,))
  9.     p.daemon = True  #
  10.     p.start()
  11.     p.join() #进程执行完毕后再关闭
  12.     print "end!"
输出:

点击(此处)折叠或打开

  1. process start: 2017-07-02 18:48:20.953754
  2. process end: 2017-07-02 18:48:25.954736

2.9 Lock()
当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突。
实例方法：
acquire([timeout]): 使线程进入同步阻塞状态，尝试获得锁定。
release(): 释放锁。使用前线程必须已获得锁定，否则将抛出异常。
例子:
多个进程使用同一个std_out ,使用lock机制确保同一个时刻有一个一个进程获取输出。

  1. #!/usr/bin/env python
    # encoding: utf-8
    """
    author: yangyi@youzan.com
    time: 2017/7/2 下午9:28
    func: 
    """
    from multiprocessing import Process, Lock
    def func_with_lock(l, i):
        l.acquire()
        print 'hello world', i
        l.release()


    def func_without_lock(i):
        print 'hello world', i


    if __name__ == '__main__':
        lock = Lock()
        print "func_with_lock :"
        for num in range(10):
            Process(target=func_with_lock, args=(lock, num)).start()


输出：

点击(此处)折叠或打开

  1. func_with_lock :
  2. hello world 0
  3. hello world 1
  4. hello world 2
  5. hello world 3
  6. hello world 4
  7. hello world 5
  8. hello world 6
  9. hello world 7
  10. hello world 8
  11. hello world 9

三 小结
 本文参考官方资料以及其他资源，对multiprocesssing 的使用方式做了总结，还有很多知识需要详细阅读官方文档。纸上来得终觉浅，绝知此事要躬行。参考资料
[1] 官方文档 
[2] Python标准库10 多进程初步 (multiprocessing包)
文章标签：
Python
Java
API
关键词：
Python multiprocessing
yangyi_dba
目录
相关文章
阿甘兄
|
9月前
|
Unix Linux Python
114 python高级 - multiprocessing
114 python高级 - multiprocessing
阿甘兄
30 0
李木子2024
|
1月前
|
数据处理 调度 Python
Python并发编程实战指南：深入理解线程（threading）与进程（multiprocessing）的奥秘，打造高效并发应用！
【7月更文挑战第8天】Python并发编程探索：使用`threading`模块创建线程处理任务，虽受限于GIL，适合I/O密集型工作。而`multiprocessing`模块通过进程实现多核利用，适用于CPU密集型任务。通过实例展示了线程和进程的创建与同步，强调了根据任务类型选择合适并发模型的重要性。
李木子2024
34 5
李木子2024
|
1月前
|
数据库 数据安全/隐私保护 C++
Python并发编程实战：线程（threading）VS进程（multiprocessing），谁才是并发之王？
【7月更文挑战第10天】Python并发对比：线程轻量级，适合I/O密集型任务，但受GIL限制；进程绕过GIL，擅CPU密集型，但通信成本高。选择取决于应用场景，线程利于数据共享，进程利于多核利用。并发无“王者”，灵活运用方为上策。
李木子2024
35 2
小空门123-30335
|
1月前
|
Python
在Python中，`multiprocessing`模块提供了一种在多个进程之间共享数据和同步的机制。
在Python中，`multiprocessing`模块提供了一种在多个进程之间共享数据和同步的机制。
小空门123-30335
24 0
小空门123-30335
|
1月前
|
安全 API Python
`multiprocessing`是Python的一个标准库，用于支持生成进程，并通过管道和队列、信号量、锁和条件变量等同步原语进行进程间通信（IPC）。
`multiprocessing`是Python的一个标准库，用于支持生成进程，并通过管道和队列、信号量、锁和条件变量等同步原语进行进程间通信（IPC）。
小空门123-30335
13 0
众所周知
|
1月前
|
API 数据库 C++
震惊！Python并发编程大揭秘：线程（threading）VS进程（multiprocessing），你选对了吗？
【7月更文挑战第8天】在Python并发编程中，线程适合I/O密集型任务，如实时订单处理，而进程适合CPU密集型任务，如商品信息同步。线程利用轻量级并发，处理I/O等待时切换成本低；进程通过multiprocessing模块充分利用多核CPU。根据任务类型选择合适工具，能提升效率并优化系统性能。理解和运用线程与进程，是解决并发问题的关键。
众所周知
20 0
didiplus
|
3月前
|
数据采集 Java Python
python并发编程:使用多进程multiprocessing模块加速程序的运行
python并发编程:使用多进程multiprocessing模块加速程序的运行
didiplus
59 1
AIweker
|
3月前
|
安全 Python
python多进程multiprocessing使用
如果你想在python中使用线程来实现并发以提高效率，大多数情况下你得到的结果是比串行执行的效率还要慢；这主要是python中GIL（全局解释锁）的缘故，通常情况下线程比较适合高IO低CPU的任务，否则创建线程的耗时可能比串行的还要多。GIL是历史问题，和C解释器有关系。 为了解决这个问题，python中提供了多进程的方式来处理需要并发的任务，可以有效的利用多核cpu达到并行的目的。【2月更文挑战第5天】
AIweker
91 0
嘿嘿就是写
|
3月前
|
并行计算 程序员 API
Python多进程编程：利用multiprocessing模块实现并行计算
Python多进程编程：利用multiprocessing模块实现并行计算
嘿嘿就是写
561 0
李木子2024
|
3月前
|
安全 Python Windows
Python 的并发编程：在 Python 中如何使用 `threading` 和 `multiprocessing` 模块？
Python 的并发编程：在 Python 中如何使用 `threading` 和 `multiprocessing` 模块？
李木子2024
62 1

热门文章

最新文章

  • 1
    深度总结 | Python编写规范
  • 2
    Aliyun Linux 2 ，误删自带python2，yum的解决办法
  • 3
    第06章节-Python3.5-Django模板语言循环字典 5
  • 4
    python基础2
  • 5
    【python】编程语言入门经典100例--25
  • 6
    Windows下Python自动补全
  • 7
    Python标准库05 存储对象 (pickle包，cPickle包)
  • 8
    Python colorama | 详解终端漂亮的彩色打印怎么实现的
  • 9
    python中__init__.py文件的作用
  • 10
    【python】编程语言入门经典100例--19
  • 1
    【Python机器学习专栏】机器学习在物联网（IoT）中的集成
    114
  • 2
    【Python 机器学习专栏】机器学习在医疗诊断中的前沿应用
    321
  • 3
    【Python机器学习专栏】联邦学习：保护隐私的机器学习新趋势
    78
  • 4
    【Python机器学习专栏】迁移学习在机器学习中的应用
    54
  • 5
    【Python 机器学习专栏】强化学习在游戏 AI 中的实践
    221
  • 6
    【Python机器学习专栏】深度学习在自动驾驶中的应用
    119
  • 7
    【Python 机器学习专栏】自然语言处理中的深度学习应用
    57
  • 8
    【Python机器学习专栏】金融数据分析中的机器学习应用
    413
  • 9
    【Python机器学习专栏】文本分类的机器学习应用
    37
  • 10
    【Python机器学习专栏】使用Python进行图像分类的实战案例
    109

    • 相关课程

    更多
  • Python Web开发基础
  • Python开发基础入门
  • Python常用数据科学库
  • Python网络爬虫实战
  • Python完全自学手册图文教程
  • Python基础快速入门实战教程

    • 相关电子书

    更多
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:
  • 双剑合璧-Python和大数据计算平台的结合

    • 相关实验场景

    更多
  • 用Python画圣诞树
  • Python选择及循环结构
  • 云端Python及基本操作
  • Python网络通信程序典型应用
  • Python新手入门（Anolis OS）
  • Python新手入门

    • 推荐镜像

    更多
  • python-release
  • nodejs-release
  • golang
    • 下一篇
    通义千问API入门教程