IT评测·应用市场-qidao123.com

标题: Python 多进程解析：Multiprocessing 高效并行处理的奥秘 [打印本页]
作者: 乌市泽哥    时间: 2024-10-19 01:14
标题: Python 多进程解析：Multiprocessing 高效并行处理的奥秘
Python 多进程解析：Multiprocessing 高效并行处理的奥秘


  
在 Python 编程中，多进程（Multiprocessing）是一种进步步伐实行服从的重要本领。本文深入解析了多进程的概念与应用，帮助开辟者充分使用多核处理器的盘算本领。我们从根本的进程创建与启动开始，解说了如何通过 Queue 实现进程间的数据传递，并通过对比多进程与多线程的性能差异，展现了多进程在处理 CPU 密集型任务时的明显优势。文章还详细先容了进程池（Pool）的使用方法，包括 map 和 apply_async 的不同应用场景。末了，我们探究了共享内存和进程锁的使用，确保多进程在并发操作中的数据安全性。本文为希望掌握多进程编程的读者提供了全面且易懂的实践指导。
一 多进程

Multiprocessing 是一种编程和实行模式，它允很多个进程同时运行，以此进步应用步伐的服从和性能。在 Python 中，multiprocessing 模块可以帮助你创建多个进程，使得每个进程都可以并行处理任务，从而有效使用多核处理器的本领。
1 导入进程标准模块

  1. import multiprocessing as mp
复制代码
2 界说调用函数

  1. def job(a, d):
  2.     print('你好 世界')
复制代码
3 创建和启动进程

  1. # 创建进程
  2. p1 = mp.Process(target=job, args=(1, 2))
  3. # 启动进程
  4. p1.start()
  5. # 连接进程
  6. p1.join()
复制代码
二 存储进程结果 Queue

1 存入输出到 Queue
  1. # 该函数没有返回值！！！
  2. def job02(q):
  3.     res = 0
  4.     for i in range(1000):
  5.         res += i + i ** 2 + i ** 3
  6.     q.put(res)  #
  8. def my_result_process02():
  9.     q = mp.Queue()
  10.     p1 = mp.Process(target=job02, args=(q,))
  11.     p2 = mp.Process(target=job02, args=(q,))
  12.     p1.start()
  13.     p2.start()
  14.     p1.join()
  15.     p2.join()
  16.     res1 = q.get()
  17.     res2 = q.get()
  18.     print(res1)
  19.     print(res2)
  20.     print(res1 + res2)
复制代码
三 threading & multiprocessing 对比

1 创建多进程 multiprocessing

  1. def job03(q):
  2.     res = 0
  3.     for i in range(1000000):
  4.         res += i + i ** 2 + i ** 3
  5.     # 结果加 queue
  6.     q.put(res)
  9. # 多核运算多进程
  10. def multicore03():
  11.     q = mp.Queue()
  12.     p1 = mp.Process(target=job03, args=(q,))
  13.     p2 = mp.Process(target=job03, args=(q,))
  14.     p1.start()
  15.     p2.start()
  16.     p1.join()
  17.     p2.join()
  18.     res1 = q.get()
  19.     res2 = q.get()
  20.     print('multicore:', res1 + res2)
复制代码
2 创建多线程 multithread

  1. # 单核运算多线程
  2. def multithread03():
  3.     # thread可放入process同样的queue中
  4.     q = mp.Queue()
  5.     t1 = td.Thread(target=job03, args=(q,))
  6.     t2 = td.Thread(target=job03, args=(q,))
  7.     t1.start()
  8.     t2.start()
  9.     t1.join()
  10.     t2.join()
  11.     res1 = q.get()
  12.     res2 = q.get()
  13.     print('multithread:', res1 + res2)
复制代码
3 创建普通函数

  1. def normal03():
  2.     res = 0
  3.     for _ in range(2):
  4.         for i in range(1000000):
  5.             res += i + i ** 2 + i ** 3
  6.     print('normal:', res)
复制代码
4 创建对比时间函数

  1. def time_result03():
  2.     st = time.time()
  3.     normal03()
  4.     st1 = time.time()
  5.     print('normal time:', st1 - st)
  6.     multithread03()
  7.     st2 = time.time()
  8.     print('multithread time:', st2 - st1)
  9.     multicore03()
  10.     print('multicore time:', time.time() - st2)
复制代码
5 运行结果

  1. normal03: 499999666667166666000000
  2. normal03 time: 0.6855959892272949
  3. multithread03: 499999666667166666000000
  4. multithread03 time: 0.6804449558258057
  5. multicore03: 499999666667166666000000
  6. multicore03 time: 0.38849496841430664
复制代码
我运行的是 normal03 > multithread03 > multicore03，normal03 和 multithread03 相差不大，multicore03 比 normal03 和 multithread03 快将近一倍。
四 进程池 Pool

使用进程池 Pool ，Python 会自行解决多进程问题。
1 进程池 Pool() 和 map()

map()  返回的是多结果。
  1. def job04(x):
  2.     # Pool的函数有返回值
  3.     return x * x
  5. def multicore04():
  6.     # Pool的函数有返回值
  7.     pool = mp.Pool()
  8.     # 自分配 CPU 计算
  9.     res = pool.map(job04, range(10))
  10.     print(res)
复制代码
2 自界说核数量

Pool 默认大小是 CPU的核数，传入 processes 参数自界说须要的核数量。
  1. def multicore05():
  2.           # 定义CPU核数量为3
  3.     pool = mp.Pool(processes=3)  
  4.     res = pool.map(job04, range(10))
  5.     print(res)
复制代码
3 apply_async 单结果返回

apply_async() 中只能传递一个值，它只会放入一个核举行运算，但是传入值时要注意是可迭代的， 所以在传入值后须要加逗号, 同时须要用 get() 方法获取返回值。
  1. def multicore06():
  2.     pool = mp.Pool()
  3.     res = pool.apply_async(job04, (2,))
  4.     # 用get获得结果
  5.     print(res.get())
复制代码
4 apply_async 多结果返回

  1. def multicore07():
  2.     pool = mp.Pool()
  3.     multi_res = [pool.apply_async(job04, (i,)) for i in range(10)]
  4.     # 用get获得结果
  5.     print([res.get() for res in multi_res])
复制代码
5 划重点


五 共享内存 shared memory

1 界说 Shared Value
  1. value1 = mp.Value('i', 0)
  2. value2 = mp.Value('d', 3.14)
复制代码
2 界说 Shared Array
它只能是一维数组
  1. array = mp.Array('i', [1, 2, 3, 4])
复制代码
其中 d 和 i 参数用来设置数据类型的，d 体现一个双精浮点类型，i 体现一个带符号的整型，参考数据类型如下：
Type codeC TypePython TypeMinimum size in bytesNotes'b'signed charint1'B'unsigned charint1'u'wchar_tUnicode character2(1)'h'signed shortint2'H'unsigned shortint2'i'signed intint2'I'unsigned intint2'l'signed longint4'L'unsigned longint4'q'signed long longint8'Q'unsigned long longint8'f'floatfloat4'd'doublefloat8 具体链接：Efficient arrays of numeric values
六 进程锁 Lock

1 不加进程锁

争抢共享内存
  1. def job08(v, num):
  2.     for _ in range(5):
  3.         time.sleep(0.1)  # 暂停0.1秒，让输出效果更明显
  4.         v.value += num  # v.value获取共享变量值
  5.         print(v.value, end="\n")
  8. def multicore08():
  9.     v = mp.Value('i', 0)  # 定义共享变量
  10.     p1 = mp.Process(target=job08, args=(v, 1))
  11.     p2 = mp.Process(target=job08, args=(v, 3))  # 设定不同的number看如何抢夺内存
  12.     p1.start()
  13.     p2.start()
  14.     p1.join()
  15.     p2.join()
复制代码
2 加进程锁

  1. def job09(v, num, l):
  2.     l.acquire()  # 锁住
  3.     for _ in range(5):
  4.         # print(v.value, num)
  5.         time.sleep(0.1)
  6.         v.value = v.value + num  # 获取共享内存
  7.         print(v.value)
  8.     l.release()  # 释放
  11. def multicore09():
  12.     l = mp.Lock()  # 定义一个进程锁
  13.     v = mp.Value('i', 0)  # 定义共享内存
  14.    
  15.     p1 = mp.Process(target=job09, args=(v, 1, l))  # 需要将lock传入
  16.     p1.start()
  17.     p1.join()
  19.     p2 = mp.Process(target=job09, args=(v, 3, l))
  20.     p2.start()
  21.     p2.join()
  24. # def multicore10():
  25. #     l = mp.Lock()  # 定义一个进程锁
  26. #     v = mp.Value('i', 0)  # 定义共享内存
  27. #     p1 = mp.Process(target=job09, args=(v, 1, l))  # 需要将lock传入
  28. #     p2 = mp.Process(target=job09, args=(v, 3, l))
  29. #     p1.start()
  30. #     p2.start()
  31. #     p1.join()
  32. #     p2.join()
复制代码
在这个示例中，必须先实行 p1 以到达预期效果。分别运行 multicore09 和 multicore10 会发现一些有意思的情况。
七 完备代码示例

注：建议在运行 main.py 对应的代码功能时，逐利用用注释举行操作。
  1. # This is a sample Python script.# Press ⌃R to execute it or replace it with your code.# Press Double ⇧ to search everywhere for classes, files, tool windows, actions, and settings.import multiprocessing as mp
  2. import threading as tdimport time as timedef print_hi(name):    # Use a breakpoint in the code line below to debug your script.    print(f'Hi, {name}')  # Press ⌘F8 to toggle the breakpoint.    # 创建进程    p1 = mp.Process(target=job, args=(1, 2))    # 启动进程    p1.start()    # Shared Value    value1 = mp.Value('i', 0)    value2 = mp.Value('d', 3.14)    # Shared Array，只能是一维数组    array = mp.Array('i', [1, 2, 3, 4])
  3. def job(a, d):
  4.     print('你好 世界')
  5. # 该函数没有返回值！！！
  6. def job02(q):
  7.     res = 0
  8.     for i in range(1000):
  9.         res += i + i ** 2 + i ** 3
  10.     q.put(res)  #
  12. def my_result_process02():
  13.     q = mp.Queue()
  14.     p1 = mp.Process(target=job02, args=(q,))
  15.     p2 = mp.Process(target=job02, args=(q,))
  16.     p1.start()
  17.     p2.start()
  18.     p1.join()
  19.     p2.join()
  20.     res1 = q.get()
  21.     res2 = q.get()
  22.     print(res1)
  23.     print(res2)
  24.     print(res1 + res2)
  25. def job03(q):    res = 0    for i in range(1000000):        res += i + i ** 2 + i ** 3    # 结果加 queue    q.put(res)# 多核运算多进程def multicore03():    q = mp.Queue()    p1 = mp.Process(target=job03, args=(q,))    p2 = mp.Process(target=job03, args=(q,))    p1.start()    p2.start()    p1.join()    p2.join()    res1 = q.get()    res2 = q.get()    print('multicore03:', res1 + res2)# 单核运算多线程def multithread03():    # thread可放入process同样的queue中    q = mp.Queue()    t1 = td.Thread(target=job03, args=(q,))    t2 = td.Thread(target=job03, args=(q,))    t1.start()    t2.start()    t1.join()    t2.join()    res1 = q.get()    res2 = q.get()    print('multithread03:', res1 + res2)def normal03():    res = 0    for _ in range(2):        for i in range(1000000):            res += i + i ** 2 + i ** 3    print('normal03:', res)def time_result03():    st = time.time()    normal03()    st1 = time.time()    print('normal03 time:', st1 - st)    multithread03()    st2 = time.time()    print('multithread03 time:', st2 - st1)    multicore03()    print('multicore03 time:', time.time() - st2)def job04(x):
  26.     # Pool的函数有返回值
  27.     return x * x
  29. def multicore04():
  30.     # Pool的函数有返回值
  31.     pool = mp.Pool()
  32.     # 自分配 CPU 计算
  33.     res = pool.map(job04, range(10))
  34.     print(res)
  35. def multicore05():    pool = mp.Pool(processes=3)  # 界说CPU核数量为3    res = pool.map(job04, range(10))    print(res)def multicore06():    pool = mp.Pool()    # apply_async() 中只能传递一个值，它只会放入一个核举行运算，但是传入值时要注意是可迭代的，    # 所以在传入值后须要加逗号, 同时须要用get()方法获取返回值    res = pool.apply_async(job04, (2,))    # 用get得到结果    print(res.get())def multicore07():
  36.     pool = mp.Pool()
  37.     multi_res = [pool.apply_async(job04, (i,)) for i in range(10)]
  38.     # 用get获得结果
  39.     print([res.get() for res in multi_res])
  40. def job08(v, num):
  41.     for _ in range(5):
  42.         time.sleep(0.1)  # 暂停0.1秒，让输出效果更明显
  43.         v.value += num  # v.value获取共享变量值
  44.         print(v.value, end="\n")
  47. def multicore08():
  48.     v = mp.Value('i', 0)  # 定义共享变量
  49.     p1 = mp.Process(target=job08, args=(v, 1))
  50.     p2 = mp.Process(target=job08, args=(v, 3))  # 设定不同的number看如何抢夺内存
  51.     p1.start()
  52.     p2.start()
  53.     p1.join()
  54.     p2.join()
  55. def job09(v, num, l):    l.acquire()  # 锁住    for _ in range(5):        # print(v.value, num)        time.sleep(0.1)        v.value = v.value + num  # 获取共享内存        print(v.value)    l.release()  # 释放def multicore09():    l = mp.Lock()  # 界说一个进程锁    v = mp.Value('i', 0)  # 界说共享内存    p1 = mp.Process(target=job09, args=(v, 1, l))  # 须要将lock传入    p1.start()    p1.join()    p2 = mp.Process(target=job09, args=(v, 3, l))    p2.start()    p2.join()def multicore10():    l = mp.Lock()  # 界说一个进程锁    v = mp.Value('i', 0)  # 界说共享内存    p1 = mp.Process(target=job09, args=(v, 1, l))  # 须要将lock传入    p2 = mp.Process(target=job09, args=(v, 3, l))    p1.start()    p2.start()    p1.join()    p2.join()# Press the green button in the gutter to run the script.if __name__ == '__main__':    print_hi('什么是 Multiprocessing')    my_result_process02()    time_result03()    multicore04()    multicore05()    multicore06()    multicore07()    multicore08()    multicore09()    # multicore10()# See PyCharm help at https://www.jetbrains.com/help/pycharm/
复制代码
复制粘贴并覆盖到你的 main.py 中运行，运行结果如下。
  1. Hi, 什么是 Multiprocessing
  2. 你好 世界
  3. 249833583000
  4. 249833583000
  5. 499667166000
  6. normal03: 499999666667166666000000
  7. normal03 time: 0.7139420509338379
  8. multithread03: 499999666667166666000000
  9. multithread03 time: 0.6696178913116455
  10. multicore03: 499999666667166666000000
  11. multicore03 time: 0.3917398452758789
  12. [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
  13. [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
  14. 4
  15. [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
  16. 3
  17. 4
  18. 7
  19. 8
  20. 11
  21. 12
  22. 1515
  24. 16
  25. 19
  26. 1
  27. 2
  28. 3
  29. 4
  30. 5
  31. 8
  32. 11
  33. 14
  34. 17
  35. 20
复制代码
八 源码地址

代码地址：
国内看 Gitee 之 什么是 Multiprocessing.py
国外看 GitHub 之 什么是 Multiprocessing.py
引用 莫烦 Python

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/) Powered by Discuz! X3.4