Python3，如何实现CPU的并行计算，那还不简单，5种方式，这篇就搞定。_python并行计算

1、引言

小屌丝：鱼哥，你上一篇《Python3，掌握这几种并行处理，轻轻松松提升for循环速度》写的很赞， 我还想继续深入并行计算。
小鱼：那你的意思，哪一篇写的不深呗？
小屌丝：我可没有那个意思。
小鱼：我也没说你是哪个意思。
小屌丝：我就是想着你能不能再讲一讲，关于如何实现CPU并行计算。
小鱼：我屮艸芔茻…你这是有啥心思？
小屌丝：我… 我没有。

小鱼：如实说来，或许，我还能讲一讲。
小屌丝：当真？
小鱼：当真…
小屌丝：就是，我要在公司分享一些技术，就涉及到 如何实现CPU并行计算方面的姿势 . 知识。
小鱼：这样啊， 那你直接说就好了。 我又不是特别喜欢黑桃A。
小屌丝：… 整，整，整吧。

2、实战

2.1 多进程模式

这里，我们主要以Python中的额多进程模式进行讲解。

2.1.1 定义

• 在Python中，多进程模式是指同时运行多个进程来执行任务的一种编程模式。
• Python提供了多个模块来支持多进程编程，其中最常用的是multiprocessing模块。

2.1.2 multiprocessing

• 使用multiprocessing模块可以创建和管理多个进程，每个进程都有自己独立的内存空间和执行环境。
• 通过将任务分配给不同的进程，可以实现并行执行，提高程序的运行效率。

2.1.3 模式

• 在多进程模式下，每个进程都有自己的主程序流程，可以独立执行任务。
• 进程之间可以通过进程间通信（IPC）机制来进行数据交换和同步操作。

2.1.4 适用场景

• 多进程模式适用于需要充分利用多核处理器或执行耗时任务的场景。
• 通过将任务分配给多个进程，可以充分利用系统资源，提高程序的运行效率。

2.1.4 代码示例

代码示例

# -*- coding:utf-8 -*-
# @Time   : 2023-07-01
# @Author : Carl_DJ


import multiprocessing

def worker(num):
    """子进程的任务函数"""
    print(f'Worker {num} started')
    # 执行一些任务
    print(f'Worker {num} finished')

if __name__ == '__main__':
    # 创建多个子进程
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start()

    # 等待所有子进程结束
    for p in processes:
        p.join()

    print('All workers finished')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

2.2 执行方法

2.2.1 多线程并发执行

多进程模式下，将任务分配给多个进程并行执行，从而利用多核CPU的优势。
这可以说作为一名码农，必备的知识点。

这里，我们同样使用multiprocessing 来实现一个并发执行任务的示例。

代码示例

# -*- coding:utf-8 -*-
# @Time   : 2023-07-01
# @Author : Carl_DJ

import multiprocessing

def task(name):
    print(f"Running task {name}")

if __name__ == "__main__":
    # 创建进程池，最大进程数为4
    pool = multiprocessing.Pool(processes=4)

    # 提交任务到进程池
    for i in range(10):
        pool.apply_async(task, args=(i,))

    # 关闭进程池，不再接受新的任务
    pool.close()

    # 等待所有任务完成
    pool.join()

    print("All tasks completed")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

解析：

• 首先创建了一个进程池，最大进程数为4；
• 其次使用apply_async方法提交了10个任务到进程池中；
• 最后，我们关闭进程池并等待所有任务完成。

2.2.2 进程池

对于大量重复的任务，
使用进程池来维护一定数量的进程，每个进程执行一个任务后返回结果，然后再由进程池分配下一个任务。
这样的好处就是：避免频繁地创建和销毁进程，从而提高效率。

我们使用 multiprocessing模块的Pool类来实现进程池。

代码示例

# -*- coding:utf-8 -*-
# @Time   : 2023-07-01
# @Author : Carl_DJ

import multiprocessing

def worker(num):
    print('Worker', num)

if __name__ == '__main__':
    # 创建一个进程池，最大进程数为3
    pool = multiprocessing.Pool(processes=3)

    # 使用进程池执行任务
    for i in range(5):
        pool.apply_async(worker, (i,))

    # 关闭进程池，不再接受新的任务
    pool.close()

    # 等待所有任务完成
    pool.join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

解析：

• 首先，创建了一个进程池，最大进程数为3；
• 其次，使用进程池的apply_async方法提交任务，每个任务都会由进程池中的一个空闲进程执行；
• 最后，关闭进程池并等待所有任务完成。

2.2.3 消息队列

在多进程模式下，不同的进程之间需要进行通信，可以利用消息队列来实现进程间通信。

我们使用Queue模块来实现消息队列。

代码示例

# -*- coding:utf-8 -*-
# @Time   : 2023-07-01
# @Author : Carl_DJ

from queue import Queue
import time

# 创建一个消息队列
message_queue = Queue()

# 生产者函数，向消息队列中添加消息
def producer():
    for i in range(5):
        message = f"Message {i+1}"
        message_queue.put(message)
        print(f"Produced: {message}")
        time.sleep(1)

# 消费者函数，从消息队列中获取消息并处理
def consumer():
    while True:
        message = message_queue.get()
        print(f"Consumed: {message}")
        time.sleep(2)
        message_queue.task_done()

# 创建并启动生产者和消费者线程
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

# 等待生产者和消费者线程结束
producer_thread.join()
consumer_thread.join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

解析：

• 使用Queue模块创建了一个消息队列message_queue；
• 生产者函数producer负责向消息队列中添加消息；
• 消费者函数consumer负责从消息队列中获取消息并进行处理。

2.2.4 内存共享

对于需要多个进程共享的数据，可以使用共享内存来避免数据拷贝和进程间通信的开销。

我们使用multiprocessing模块的Value和Array类来实现共享内存

代码示例

# -*- coding:utf-8 -*-
# @Time   : 2023-07-01
# @Author : Carl_DJ

from multiprocessing import Process, Value, Array

# 定义一个共享变量
shared_value = Value('i', 0)

# 定义一个共享数组
shared_array = Array('d', [0.0, 1.0, 2.0, 3.0, 4.0])

# 定义一个函数，用于修改共享变量和数组的值
def modify_shared_data(value, array):
    value.value = 10
    for i in range(len(array)):
        array[i] = i * 2

# 创建一个子进程，传入共享变量和数组
p = Process(target=modify_shared_data, args=(shared_value, shared_array))
p.start()
p.join()

# 打印共享变量和数组的值
print("Shared value:", shared_value.value)
print("Shared array:", shared_array[:])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

解析：

• 首先，使用Value和Array类分别创建了一个共享变量shared_value和一个共享数组shared_array；
• 其次，定义了一个函数modify_shared_data，该函数会修改共享变量和数组的值；
• 第三，创建了一个子进程，将共享变量和数组作为参数传递给子进程；
• 最后，子进程执行完毕后，输出共享变量和数组的值。

2.2.5 异步IO

对于I/O密集型任务，可以使用异步IO来提高效率。

我们使用asyncio模块来实现异步IO。

代码示例

# -*- coding:utf-8 -*-
# @Time   : 2023-07-01
# @Author : Carl_DJ


import asyncio

async def fetch_data(url):
    print(f"正在请求URL：{url}")
    await asyncio.sleep(2)  # 模拟网络请求延迟
    print(f"请求URL：{url}完成")
    return f"从{url}获取的数据"

async def main():
    urls = [
        "https://www.example.com",
        "https://www.google.com",
        "https://www.baidu.com"
    ]
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    print(results)

if __name__ == "__main__":
    asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

解析：

• 首先，定义一个fetch_data函数，模拟了一个网络请求，并使用asyncio.sleep来模拟请求的延迟；
• 其次，在main函数中，创建了多个fetch_data的协程任务，并使用asyncio.gather来并发执行这些任务；
• 最后，使用asyncio.run来运行main函数。

3、总结

看到这里，今天的分享差不多就到这里了。
今天主要针对在Python中， 使用多进程模式来实现CPU的并行计算。如：

• 多线程并发执行
• 进程池
• 消息队列
• 内存共享
• 异步IO

在实际的项目中，很多地方都会用到并行计算， 这不仅提高的代码执行效率， 也提高了用户的满意度。

我是小鱼：

• CSDN 博客专家；
• 阿里云 专家博主；
• 51CTO博客专家；
• 51认证讲师等；
• 认证金牌面试官；
• 职场面试及培训规划师；
• 多个国内主流技术社区的认证专家博主；
• 多款主流产品(阿里云等)测评一、二等奖获得者；

关注我，带你学习更多更专业更前言的Python技术。