首页 > 脚本专栏 > python > Python multiprocessing多线程

Python使用multiprocessing实现多进程

2024-10-12 11:40:27 作者：寒秋丶

当我们工作中涉及到处理大量数据、并行计算或并发任务时,Python的multiprocessing模块是一个强大而实用的工具,在本文中,我们将探索如何使用multiprocessing模块实现多进程编程,将介绍进程池的概念和用法,需要的朋友可以参考下

引言

大家好，当我们工作中涉及到处理大量数据、并行计算或并发任务时，Python的multiprocessing模块是一个强大而实用的工具。通过它，我们可以轻松地利用多核处理器的优势，将任务分配给多个进程并同时执行，从而提高程序的性能和效率。在本文中，我们将探索如何使用multiprocessing模块实现多进程编程。将介绍进程池的概念和用法，以及如何使用它来管理和调度多个进程。我们还将讨论并发任务的处理、进程间通信和结果获取等关键问题，希望能给大家的工作带来一些帮助。

一、介绍

Python多进程是一种并行编程模型，允许在Python程序中同时执行多个进程。每个进程都拥有自己的独立内存空间和执行环境，可以并行地执行任务，从而提高程序的性能和效率。

优点：

并行处理：多进程可以同时执行多个任务，充分利用多核处理器的能力，实现并行处理。这可以显著提高程序的性能和效率，特别是在处理密集型任务或需要大量计算的场景中。
独立性：每个进程都有自己的独立地址空间和执行环境，进程之间互不干扰。这意味着每个进程都可以独立地执行任务，不会受到其他进程的影响。这种独立性使得多进程编程更加健壮和可靠。
内存隔离：由于每个进程都拥有自己的地址空间，多进程之间的数据是相互隔离的。这意味着不同进程之间的变量和数据不会相互影响，减少了数据共享和同步的复杂性。
故障隔离：如果一个进程崩溃或出现错误，不会影响其他进程的执行。每个进程是独立的实体，一个进程的故障不会对整个程序产生致命影响，提高了程序的稳定性和容错性。
可移植性：多进程编程可以在不同的操作系统上运行，因为进程是操作系统提供的基本概念。这使得多进程编程具有很好的可移植性，可以在不同的平台上部署和运行。

缺点：

资源消耗：每个进程都需要独立的内存空间和系统资源，包括打开的文件、网络连接等。多进程编程可能会增加系统的资源消耗，尤其是在创建大量进程时。
上下文切换开销：在多进程编程中，进程之间的切换需要保存和恢复进程的执行环境，这涉及到上下文切换的开销。频繁的进程切换可能会导致额外的开销，影响程序的性能。
数据共享与同步：由于多进程之间的数据是相互隔离的，需要通过特定的机制进行数据共享和同步。这可能涉及到进程间通信（IPC）的复杂性，如队列、管道、共享内存等。正确处理数据共享和同步是多进程编程中的挑战之一。
编程复杂性：相比于单线程或多线程编程，多进程编程可能更加复杂。需要考虑进程的创建和管理、进程间通信、数据共享和同步等问题。编写和调试多进程程序可能需要更多的工作和经验。

进程与线程：

在讨论多进程之前，需要明确进程（Process）和线程（Thread）的概念。
进程是计算机中正在运行的程序的实例。每个进程都有自己的地址空间、数据栈和控制信息，可以独立执行任务。
线程是进程中的一个执行单元，可以看作是轻量级的进程。多个线程共享同一进程的资源，包括内存空间、文件描述符等。

多进程编程在并行处理和资源隔离方面具有明显的优势，但也涉及到资源消耗、上下文切换开销、数据共享和同步等问题。在实际应用中，开发者应权衡利弊，根据具体场景选择适合的编程模型和工具。

二、创建进程

在Python中，可以使用multiprocessing模块来创建和管理进程。该模块提供了丰富的类和函数，用于创建、启动和管理进程。

1、导入multiprocessing模块

在使用multiprocessing模块之前，需要先导入它：

import multiprocessing

2、创建进程

可以使用multiprocessing.Process类来创建进程对象。需要传入一个目标函数作为进程的执行逻辑。可以通过继承multiprocessing.Process类来自定义进程类。

import multiprocessing
 
def worker():
    # 进程执行的逻辑
 
if __name__ == '__main__':
    process = multiprocessing.Process(target=worker)

在上面的示例中，worker函数是进程的执行逻辑。进程对象创建后，可以通过设置参数、调用方法等来配置进程。

3、启动进程

通过调用进程对象的start()方法，可以启动进程。进程会在后台开始执行。

process.start()

4、进程的状态

进程对象提供了一些方法来获取和管理进程的状态：

is_alive()：检查进程是否正在运行。
join([timeout])：等待进程结束。可选参数timeout指定等待的最长时间。

if process.is_alive():
    print("进程正在运行")
 
process.join()

二、进程间通信

进程间通信（Inter-Process Communication，IPC）是指不同进程之间进行数据交换和共享信息的机制。在多进程编程中，进程之间通常需要进行数据传输、共享状态或进行同步操作。Python提供了多种进程间通信的机制，包括队列（Queue）、管道（Pipe）、共享内存（Value、Array）等。

1、队列（Queue）

队列是一种常用的进程间通信方式，通过队列可以实现进程之间的数据传输。Python的multiprocessing模块提供了Queue类来实现多进程之间的队列通信。进程可以通过put()方法将数据放入队列，其他进程则可以通过get()方法从队列中获取数据。

from multiprocessing import Queue
 
# 创建队列
queue = Queue()
 
# 进程1放入数据
queue.put(data)
 
# 进程2获取数据
data = queue.get()

2、管道（Pipe）

管道是另一种常用的进程间通信方式，通过管道可以实现进程之间的双向通信。Python的multiprocessing模块提供了Pipe类来创建管道对象。Pipe()方法返回两个连接的管道端，一个用于发送数据，另一个用于接收数据。

from multiprocessing import Pipe
 
# 创建管道
conn1, conn2 = Pipe()
 
# 进程1发送数据
conn1.send(data)
 
# 进程2接收数据
data = conn2.recv()

3、共享内存（Value、Array）

共享内存是一种在多进程之间共享数据的高效方式。Python的multiprocessing模块提供了Value和Array类来实现进程间共享数据。Value用于共享单个值，而Array用于共享数组。

from multiprocessing import Value, Array
 
# 创建共享值
shared_value = Value('i', 0)
 
# 创建共享数组
shared_array = Array('i', [1, 2, 3, 4, 5])

在创建共享值和共享数组时，需要指定数据类型（如整数、浮点数）和初始值。进程可以通过读写共享值和共享数组来进行进程间的数据共享。

4、信号量（Semaphore）

信号量是一种用于控制对共享资源的访问的机制。在多进程编程中，信号量可以用于限制同时访问某个共享资源的进程数量。

from multiprocessing import Semaphore, Process
import time
 
def worker(semaphore, name):
    semaphore.acquire()
    print("Worker", name, "acquired semaphore")
    time.sleep(2)
    print("Worker", name, "released semaphore")
    semaphore.release()
 
semaphore = Semaphore(2)
 
processes = []
for i in range(5):
    p = Process(target=worker, args=(semaphore, i))
    processes.append(p)
    p.start()
 
for p in processes:
    p.join()

在上述例子中，创建了一个信号量，初始值为2。然后创建了5个进程，每个进程在执行前会尝试获取信号量，如果信号量的值大于0，则成功获取；否则，进程将被阻塞，直到有进程释放信号量。每个进程获取信号量后，会执行一段任务，并在执行完后释放信号量。

5、事件（Event）

事件是一种用于多进程间通信的同步机制，它允许一个或多个进程等待某个事件的发生，然后再继续执行。

from multiprocessing import Event, Process
import time
 
def worker(event, name):
    print("Worker", name, "waiting for event")
    event.wait()
    print("Worker", name, "received event")
    time.sleep(2)
    print("Worker", name, "completed task")
 
event = Event()
 
processes = []
for i in range(3):
    p = Process(target=worker, args=(event, i))
    processes.append(p)
    p.start()
 
time.sleep(3)
event.set()
 
for p in processes:
    p.join()

在上述例子中，创建了一个事件。然后创建了3个进程，每个进程在执行前会等待事件的发生，即调用event.wait()方法。主进程休眠3秒后，设置事件的状态为已发生，即调用event.set()方法。此时，所有等待事件的进程将被唤醒，并继续执行任务。

6、条件变量（Condition）

条件变量是一种用于多进程间协调和同步的机制，它可以用于控制多个进程之间的执行顺序。

from multiprocessing import Condition, Process
import time
 
def consumer(condition):
    with condition:
        print("Consumer is waiting")
        condition.wait()
        print("Consumer is consuming the product")
 
def producer(condition):
    with condition:
        time.sleep(2)
        print("Producer is producing the product")
        condition.notify()
 
condition = Condition()
 
consumer_process = Process(target=consumer, args=(condition,))
producer_process = Process(target=producer, args=(condition,))
 
consumer_process.start()
producer_process.start()
 
consumer_process.join()
producer_process.join()

在上述例子中，创建了一个条件变量。然后创建了一个消费者进程和一个生产者进程。消费者进程在执行前等待条件的满足，即调用condition.wait()方法。生产者进程休眠2秒后，生成产品并通过condition.notify()方法通知消费者。消费者收到通知后继续执行任务。

三、进程间同步

进程间同步是确保多个进程按照特定顺序执行或在共享资源上进行互斥访问的一种机制。进程间同步的目的是避免竞态条件（race condition）和数据不一致的问题。Python提供了多种机制来实现进程间的同步，包括锁（Lock）、信号量（Semaphore）、事件（Event）、条件变量（Condition）等。

1、锁（Lock）

锁是一种最基本的同步机制，用于保护共享资源的互斥访问，确保在任意时刻只有一个进程可以访问共享资源。在Python中，可以使用multiprocessing模块的Lock类来实现锁。

from multiprocessing import Lock, Process
 
lock = Lock()
 
def worker(lock, data):
    lock.acquire()
    try:
        # 对共享资源进行操作
        pass
    finally:
        lock.release()
 
processes = []
for i in range(5):
    p = Process(target=worker, args=(lock, i))
    processes.append(p)
    p.start()
 
for p in processes:
    p.join()

在上述例子中，每个进程在访问共享资源之前会先获取锁，然后在完成操作后释放锁。这样可以确保在同一时刻只有一个进程能够访问共享资源，避免数据竞争问题。

2、信号量（Semaphore）

信号量是一种更为灵活的同步机制，它允许多个进程同时访问某个资源，但限制同时访问的进程数量。在Python中，可以使用multiprocessing模块的Semaphore类来实现信号量。

from multiprocessing import Semaphore, Process
 
semaphore = Semaphore(2)
 
def worker(semaphore, data):
    semaphore.acquire()
    try:
        # 对共享资源进行操作
        pass
    finally:
        semaphore.release()
 
processes = []
for i in range(5):
    p = Process(target=worker, args=(semaphore, i))
    processes.append(p)
    p.start()
 
for p in processes:
    p.join()

在上述例子中，创建了一个初始值为2的信号量。每个进程在访问共享资源之前会尝试获取信号量，只有当信号量的值大于0时才能获取成功，否则进程将被阻塞。获取成功后，进程可以进行操作，并在完成后释放信号量。

3、事件（Event）

事件是一种同步机制，用于实现进程之间的等待和通知机制。一个进程可以等待事件的发生，而另一个进程可以触发事件的发生。在Python中，可以使用multiprocessing模块的Event类来实现事件。

from multiprocessing import Event, Process
 
event = Event()
 
def worker(event, data):
    event.wait()
    # 执行任务
 
processes = []
for i in range(5):
    p = Process(target=worker, args=(event, i))
    processes.append(p)
    p.start()
 
# 触发事件的发生
event.set()
 
for p in processes:
    p.join()

在上述例子中，多个进程在执行任务前会等待事件的发生，即调用event.wait()方法。主进程通过调用event.set()方法来触发事件的发生，进而唤醒等待的进程继续执行。

4、条件变量（Condition）

条件变量是一种复杂的同步机制，它允许进程按照特定的条件等待和通知。在Python中，可以使用multiprocessing模块的Condition类来实现条件变量。

from multiprocessing import Condition, Process
 
condition = Condition()
 
def consumer(condition（续）：
def consumer(condition, data):
    with condition:
        while True:
            # 检查条件是否满足
            while not condition_is_met():
                condition.wait()
            # 从共享资源中消费数据
def producer(condition, data):
    with condition:
        # 生成数据并更新共享资源
        condition.notify_all()
processes = []
for i in range(5):
    p = Process(target=consumer, args=(condition, i))
    processes.append(p)
    p.start()
producer_process = Process(target=producer, args=(condition, data))
producer_process.start()
for p in processes:
    p.join()
producer_process.join()

在上述例子中，消费者进程在执行任务前会检查条件是否满足，如果条件不满足，则调用condition.wait()方法等待条件的满足。生产者进程生成数据并更新共享资源后，调用condition.notify_all()方法通知所有等待的消费者进程条件已满足。被唤醒的消费者进程会重新检查条件并执行任务。

四、进程池

进程池是一种用于管理和调度多个进程的机制，它可以有效地处理并行任务和提高程序的性能。进程池在Python中通常使用multiprocessing模块提供的Pool类来实现。

进程池的工作原理如下：

创建进程池时，会启动指定数量的进程，并将它们放入池中。
池中的进程会等待主进程提交任务。
主进程通过提交任务给进程池，将任务分配给空闲的进程。
进程池中的进程执行任务，并将结果返回给主进程。
主进程获取任务的结果，继续执行其他操作。
当所有任务完成后，主进程关闭进程池。

1、创建进程池

要使用进程池，首先需要创建一个Pool对象，可以指定池中的进程数量。通常，可以使用multiprocessing.cpu_count()函数来获取当前系统的CPU核心数，然后根据需要来指定进程池的大小。

from multiprocessing import Pool, cpu_count
 
pool = Pool(processes=cpu_count())

在上述例子中，创建了一个进程池，进程数量与系统的CPU核心数相同。

2、提交任务

一旦创建了进程池，就可以使用apply()、map()或imap()方法来提交任务给进程池。

apply()方法用于提交单个任务，并等待任务完成后返回结果。

result = pool.apply(function, args=(arg1, arg2))

map()方法用于提交多个任务，并按照任务提交的顺序返回结果列表。

results = pool.map(function, iterable)

imap()方法也用于提交多个任务，但可以通过迭代器逐个获取结果，而不需要等待所有任务完成。

results = pool.imap(function, iterable)

在上述例子中，function表示要执行的函数，args是函数的参数，iterable是一个可迭代对象，可以是列表、元组等。

3、获取结果

对于apply()方法，调用后会阻塞主进程，直到任务完成并返回结果。对于map()方法，调用后会等待所有任务完成，并按照任务提交的顺序返回结果列表。对于imap()方法，可以通过迭代器逐个获取结果。

for result in results:
    print(result)

在上述例子中，使用for循环逐个获取结果并进行处理。

4、关闭进程池

在所有任务完成后，需要显式地关闭进程池，以释放资源。

pool.close()
pool.join()

调用close()方法后，进程池将不再接受新的任务。调用join()方法会阻塞主进程，直到所有任务都已完成。

5、使用进程池的示例

from multiprocessing import Pool
 
# 定义一个任务函数
def square(x):
    return x ** 2
 
if __name__ == '__main__':
    # 创建进程池
    with Pool(processes=4) as pool:
        # 提交任务给进程池
        results = pool.map(square, range(10))
 
    # 打印结果
    print(results)

在上述示例中，首先定义了一个任务函数square，它接受一个数值作为参数，并返回该数值的平方。

在if __name__ == '__main__':中，创建了一个进程池，指定进程数量为4。使用with语句可以确保进程池在使用完毕后被正确关闭。

然后，通过pool.map(square, range(10))将任务提交给进程池。map()方法会将任务函数square和一个可迭代对象range(10)作为参数，它会将可迭代对象中的每个元素依次传递给任务函数进行处理，并返回结果列表。最后，打印结果列表，即每个数值的平方。

需要注意的是，在使用进程池时，需要将主程序代码放在if __name__ == '__main__':中，以确保在子进程中不会重复执行主程序的代码。

以下是一个更加复杂的多进程示例，展示了如何使用进程池处理多个任务，并在任务完成时获取结果。

import time
from multiprocessing import Pool
 
# 定义一个任务函数
def process_data(data):
    # 模拟耗时操作
    time.sleep(1)
    # 返回处理结果
    return data.upper()
 
if __name__ == '__main__':
    # 创建进程池
    with Pool(processes=3) as pool:
        # 准备数据
        data_list = ['apple', 'banana', 'cherry', 'date', 'elderberry']
 
        # 提交任务给进程池
        results = [pool.apply_async(process_data, args=(data,)) for data in data_list]
 
        # 等待所有任务完成并获取结果
        final_results = [result.get() for result in results]
 
    # 打印结果
    for result in final_results:
        print(result)

在上述示例中，除了使用进程池的map()方法提交任务之外，还使用了apply_async()方法来异步提交任务，并通过get()方法获取任务的结果。

在if __name__ == '__main__':中，创建了一个进程池，指定进程数量为3。使用with语句可以确保进程池在使用完毕后被正确关闭。然后，准备了一个数据列表data_list，其中包含了需要处理的数据。

通过列表推导式，使用pool.apply_async(process_data, args=(data,))将任务异步提交给进程池。apply_async()方法会将任务函数process_data和数据data作为参数，返回一个AsyncResult对象，表示异步任务的结果。将这些对象存储在results列表中。

接下来，使用列表推导式，通过result.get()方法等待所有任务完成并获取结果，将结果存储在final_results列表中。最后，使用for循环遍历final_results列表，并打印每个任务的处理结果。

进程池的优点是可以自动管理和调度多个进程，充分利用系统资源，提高程序的并行执行能力。通过合理设置进程池的大小，可以在不过度消耗系统资源的情况下，实现最佳的并发效果。但需要注意的是，进程池适用于那些需要并行执行的任务，而不适用于IO密集型任务，因为进程池中的进程是通过复制主进程来创建的，而IO密集型任务更适合使用线程池来实现并发。

以上就是Python使用multiprocessing实现多进程的详细内容，更多关于Python multiprocessing多线程的资料请关注脚本之家其它相关文章！