一文学会python多线程

Python的 multiprocessing 模块是一个多进程模块，它允许你创建新的进程，就像你创建新的线程一样。

Python的 multiprocessing 模块提供了一个Process类 来代表一个进程对象。multiprocessing模块的API和threading模块非常相似，所以如果你熟悉线程的编程，那么你会发现使用multiprocessing模块创建多进程是很直观的。

创建的新进程有自己的Python解释器和内存空间，因此它们可以并行运行，并且不受全局解释器锁（GIL）的限制，这使得multiprocessing模块对于CPU密集型任务特别有效。

同时，multiprocessing模块还提供了一些强大的功能，比如进程间的通信、共享状态、进程池等等，这使得我们可以更好地管理和控制多进程的行为。

（一）先来个简单的例子：

正常我们打印一个list的值是按顺序的：

def process_data(i):
    print(f'Process {i} is processing data.')

if __name__ == "__main__":
    data_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 

    for i in data_list:
        process_data(i)

运行结果：

Process 1 is processing data.
Process 2 is processing data.
Process 3 is processing data.
Process 4 is processing data.
Process 5 is processing data.
Process 6 is processing data.
Process 7 is processing data.
Process 8 is processing data.
Process 9 is processing data.
Process 10 is processing data.

使用多线程实现打印：使用multiprocessing模块来创建多进程。

下面是一个示例，在此示例中，每个进程都会处理列表中的一个元素：

from multiprocessing import Process

def process_data(i):
    print(f'Process {i} is processing data.')

if __name__ == "__main__":
    data_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 

    processes = []

    for i in data_list:
        p = Process(target=process_data, args=(i,))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

运行结果：

Process 1 is processing data.
Process 2 is processing data.
Process 3 is processing data.
Process 4 is processing data.
Process 5 is processing data.
Process 6 is processing data.
Process 7 is processing data.
Process 8 is processing data.
Process 9 is processing data.
Process 10 is processing data.

需要说明一下：join()方法是Python中multiprocessing和threading模块的一部分。在多进程和多线程编程中，join()方法的主要功能是等待进程或线程完成。

当你在一个进程或线程对象上调用join()方法时，调用它的线程（通常是主线程）会被阻塞，直到被join()的进程或线程完成。

在多进程或多线程编程中，join()方法很重要，因为它可以确保在主线程（或任何其他线程）中正确地等待其他线程或进程完成。这通常对于同步操作，以及在退出程序之前清理和完成所有的并行操作，是必要的。

在你给出的代码中，join()方法被用于确保所有的进程都完成了它们的工作，然后主进程（运行这段代码的进程）可以继续进行。

（二）

言归正传，上面的代码看不出来是否是同时进行的，那么我们加一个sleep时间进去，让每个线程运行时先睡一觉：

from multiprocessing import Process
import time
import random

def process_data(i, time_to_sleep):
    time.sleep(time_to_sleep)
    print(f'Process {i} is processing data, sleeping for {time_to_sleep} seconds.')

if __name__ == "__main__":
    data_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 

    processes = []

    for i in data_list:
        time_to_sleep = random.randint(1, 5)
        p = Process(target=process_data, args=(i,time_to_sleep,))
        processes.append(p)
        p.start()

    for process in processes:
        process.join()

运行结果：

Process 2 is processing data, sleeping for 2 seconds.
Process 7 is processing data, sleeping for 2 seconds.
Process 3 is processing data, sleeping for 2 seconds.
Process 6 is processing data, sleeping for 2 seconds.
Process 1 is processing data, sleeping for 3 seconds.
Process 9 is processing data, sleeping for 4 seconds.
Process 4 is processing data, sleeping for 4 seconds.
Process 5 is processing data, sleeping for 4 seconds.
Process 10 is processing data, sleeping for 5 seconds.
Process 8 is processing data, sleeping for 5 seconds.

这样就能看到，睡的时间短的线程先完成。

（三）

接下来，我们把cpu 数加进去，让它在规定数量的cpu中运行，一次启动等于CPU核心数量的进程，然后当这些进程完成时启动新的进程。这样可以确保并行执行的进程数量不超过CPU核心数量。

这时就要引进queue（队列），queue被用来存储待处理的数据。queue的特点是先入先出（FIFO，First In First Out），这使得我们能按照数据放入queue的顺序依次处理数据。

在这个例子中，data_list中的所有元素都被放入queue中，然后在新的进程启动时从queue中取出。这个方法的好处是，一旦所有的数据都被放入queue，我们就可以不用关心数据是怎么被分配到各个进程的，只需要在每个进程中从queue中取出数据并处理就可以了。

此外，Python的queue模块提供了一种线程安全的数据结构，适用于多线程或多进程之间的数据交换，可以避免因并发访问而引起的数据不一致问题。

from multiprocessing import Process
import multiprocessing
import queue
import time
import random


def process_data(data, time_to_sleep):
    time.sleep(time_to_sleep)
    print(f'Processing {data} is processing data, sleeping for {time_to_sleep} seconds')


if __name__ == '__main__':
    data_list = [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15]

    processes = []

    cpu_cores = multiprocessing.cpu_count()

    # Create a Queue to hold the data_list
    data_queue = queue.Queue()
    for data in data_list:
        data_queue.put(data)

    while not data_queue.empty():
        # Start new processes until we reach cpu_cores
        while len(processes) < cpu_cores and not data_queue.empty():
            data = data_queue.get()
            # sleep random time
            time_to_sleep = random.randint(1,5)
            # time_to_sleep = 1
            p = Process(target=process_data, args=(data,time_to_sleep,))
            p.start()
            processes.append(p)

        # Remove and join any processes that have finished
        processes = [p for p in processes if p.is_alive()]
    
    # Wait for all processes to finish
    for process in processes:
        process.join()

    print('Finished Processing')

但需要注意的是，上述代码中的queue是Python的内置queue，它并不支持进程间的通信。如果你需要在不同的进程之间共享数据，你应该使用multiprocessing模块中的Queue，它是设计用来在进程间安全交换信息的。因为这个例子中没有用到进程间的数据交换，就没有使用multiprocessing模块中的Queue。

在这个版本的程序中，我们首先创建一个队列来保存所有的数据，并且一直运行直到队列为空。在每次循环中，我们开始新的进程直到达到cpu_cores数量，并且从队列中移除已完成的进程。当队列为空时，我们仍然会等待所有剩余的进程完成。

这里还可以将cpu_cores手动设置为1，再看看打印效果是不是跟单线程的效果一样（本文不再测试，读者感兴趣的自己测试）

（四）

接下来加大难度，通常情况下我们的data_process函数一般是要返回一个值的，并且在主程序中需要接收这个返回值。

那么如何接收这个值呢？假设我们每个函数都要返回一个字典，在所有的进程结束后，主程序将每个子进程的字典合并成一个大字典。

使用 multiprocessing.Pool 可以更轻松地收集进程返回的结果。 Pool.map 可以自动收集每个进程的结果，并将这些结果作为一个列表返回。下面是一个示例：

from multiprocessing import Pool, cpu_count
import random
import time

def process_data(data):
    time_to_sleep = random.randint(1, 5)
    time.sleep(time_to_sleep)
    print(f'Processing {data} is processing data, sleeping for {time_to_sleep} seconds')
    return {data: time_to_sleep}


if __name__ == "__main__":
    data_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    
    cpu_cores = cpu_count() # 获取 CPU 核心数

    # 使用 Pool 来创建进程池
    with Pool(cpu_cores) as pool:
        results = pool.map(process_data, data_list )  # 使用 Pool.map 来执行进程并收集结果

    # 将结果合并为一个大字典
    combined_dict = {}
    for result in results:
        combined_dict.update(result)

    print(combined_dict)  # 打印出合并后的大字典

输出结果：

Processing 1 is processing data, sleeping for 1 seconds
Processing 4 is processing data, sleeping for 1 seconds
Processing 7 is processing data, sleeping for 1 seconds
Processing 3 is processing data, sleeping for 2 seconds
Processing 9 is processing data, sleeping for 1 seconds
Processing 6 is processing data, sleeping for 2 seconds
Processing 10 is processing data, sleeping for 2 seconds
Processing 2 is processing data, sleeping for 4 seconds
Processing 8 is processing data, sleeping for 5 seconds
Processing 5 is processing data, sleeping for 5 seconds
{1: 1, 2: 4, 3: 2, 4: 1, 5: 5, 6: 2, 7: 1, 8: 5, 9: 1, 10: 2}

说明：对于multiprocessing.Pool().map()方法，它的设计是这样的：第一个参数是一个函数，第二个参数是一个可迭代对象，其中每个元素都会被作为第一个参数函数的参数。但是，Pool().map()并不能直接处理需要多个参数的函数。对于多参数函数，我们可以使用Pool().starmap()方法，该方法的参数是一个可迭代对象，每一个元素是一个元组，元组的元素会作为函数的参数。

因此，假设你需要给process_data函数传入两个参数，那么你可以这样修改代码：

from multiprocessing import Pool, cpu_count
import time
import random

def process_data(data, time_to_sleep):
    time.sleep(time_to_sleep)
    print(f'Processing {data} is processing data, sleeping for {time_to_sleep} seconds')
    return {data: time_to_sleep}

if __name__ == "__main__":
    data_list = [(i, random.randint(1, 5)) for i in range(1, 11)]  # 创建一个元组列表，每个元组包含两个参数

    cpu_cores = cpu_count()

    with Pool(cpu_cores) as pool:
        results = pool.starmap(process_data, data_list)  # 使用 Pool.starmap() 来执行进程并收集结果

    # 将结果合并为一个大字典
    combined_dict = {}
    for result in results:
        combined_dict.update(result)

    print(combined_dict)  # 打印出合并后的大字典

在这个修改后的例子中，我们首先创建了一个包含元组的列表data_list，其中每个元组包含两个元素，分别是process_data函数需要的两个参数。然后我们使用pool.starmap()函数来处理这个列表，它将会把列表中的每一个元组的元素拆解出来，作为process_data函数的参数。

总结：这次你学会了如果使用python的多线程了吧，根据自己的应用场景，套上面的例子即可。

• cpu_cores的获取：multiprocessing.cpu_count
• 使用 Pool.map 来执行进程并收集结果
• 有参数的和无参数的函数都适用