首页　>　文章列表　>　Python 中的并发性与线程和多处理

Python 中的并发性与线程和多处理

329 2024-09-25

Python 中的并发性与线程和多处理

并发是现代编程中的一个重要思想，它允许多个任务同时运行以提高应用程序的性能。

在 python 中实现并发的方法有多种，其中最著名的是线程和多处理。

在本文中，我们将详细探讨这两种方法，了解它们的工作原理，并讨论何时使用每种方法，以及实际的代码示例。

什么是并发？

在我们讨论线程和多处理之前，了解并发的含义很重要。

并发是指一个程序可以同时执行多个任务或进程。

这可以使程序更好地利用资源并运行得更快，特别是当它需要执行诸如读取文件或进行大量计算之类的操作时。

实现并发的方式主要有两种：

并行性：在计算机处理器的不同部分上同时运行多个任务。
并发：在同一时间段内处理多个任务，但不一定在完全相同的时刻。

python 提供了两种主要方式来实现并发：

线程：适用于可以同时管理的任务。
多处理：适用于需要在不同处理器核心上真正同时运行的任务。

python 中的线程

线程允许您在同一进程内运行多个较小的进程单元（称为线程），共享相同的内存空间。

线程比进程更轻，并且它们之间的切换更快。

但是，python 中的线程受全局解释器锁 (gil) 的约束，这确保一次只有一个线程可以执行 python 代码。

线程如何工作

python的线程模块提供了一种简单灵活的方式来创建和管理线程。

让我们从一个基本示例开始：

import threading
import time


def print_numbers():
    for i in range(5):
        print(f"number: {i}")
        time.sleep(1)


# creating a thread
thread = threading.thread(target=print_numbers)

# starting the thread
thread.start()

# wait for the thread to complete
thread.join()

print("thread has finished executing")


# output:
# number: 0
# number: 1
# number: 2
# number: 3
# number: 4
# thread has finished executing

在此示例中：

我们定义了一个函数 print_numbers()，它打印从 0 到 4 的数字，两次打印之间有一秒的延迟。
我们使用 threading.thread() 创建一个线程，并将 print_numbers() 作为目标函数传递。
start() 方法开始线程的执行，而 join() 确保主程序等待线程完成后再继续执行。

示例：i/o 密集型任务的线程化

线程对于 i/o 密集型任务特别有用，例如文件操作、网络请求或数据库查询，在这些任务中程序大部分时间都在等待外部资源。

这是一个使用线程模拟下载文件的示例：

import threading
import time


def download_file(file_name):
    print(f"starting download of {file_name}...")
    time.sleep(2)  # simulate download time
    print(f"finished downloading {file_name}")


files = ["file1.zip", "file2.zip", "file3.zip"]

threads = []

# create and start threads
for file in files:
    thread = threading.thread(target=download_file, args=(file,))
    thread.start()
    threads.append(thread)

# ensure all threads have finished
for thread in threads:
    thread.join()

print("all files have been downloaded.")

# output:
# starting download of file1.zip...
# starting download of file2.zip...
# starting download of file3.zip...
# finished downloading file1.zip
# finished downloading file2.zip
# finished downloading file3.zip
# all files have been downloaded.

通过为每个文件下载创建和管理单独的线程，程序可以同时处理多个任务，从而提高整体效率。

代码中关键步骤如下：

定义了一个函数 download_file 来模拟下载过程。
创建文件名列表来表示需要下载的文件。
对于列表中的每个文件，都会创建一个新线程，并以 download_file 作为其目标函数。每个线程在创建后立即启动并添加到线程列表中。
主程序使用 join() 方法等待所有线程完成，确保程序在所有下载完成之前不会继续进行。

线程的局限性

虽然线程可以提高 i/o 密集型任务的性能，但它也有局限性：

全局解释器锁 (gil)：对于 cpu 密集型任务，gil 限制一次只能执行一个线程，从而限制了多核处理器中线程的有效性。
竞争条件：由于线程共享相同的内存空间，不正确的同步可能会导致竞争条件，程序的结果取决于线程的时间。
死锁：线程相互等待释放资源可能会导致死锁，从而无法取得任何进展。

python 中的多处理

多处理通过使用单独的进程而不是线程来解决线程的局限性。

每个进程都有自己的内存空间和python解释器，允许在多核系统上实现真正的并行。

这使得多重处理成为需要大量计算的任务的理想选择。

多重处理的工作原理

python 中的多处理模块允许您轻松创建和管理进程。

让我们从一个基本示例开始：

import multiprocessing
import time


def print_numbers():
    for i in range(5):
        print(f"number: {i}")
        time.sleep(1)


if __name__ == "__main__":
    # creating a process
    process = multiprocessing.process(target=print_numbers)

    # starting the process
    process.start()

    # wait for the process to complete
    process.join()

    print("process has finished executing")

# output:
# number: 0
# number: 1
# number: 2
# number: 3
# number: 4
# process has finished executing

此示例与线程示例类似，但具有进程。

请注意，进程的创建和管理与线程类似，但由于进程运行在单独的内存空间中，因此它们是真正并发的，并且可以运行在不同的 cpu 核心上。

示例：cpu 密集型任务的多处理

多处理对于受 cpu 限制的任务特别有用，例如数值计算或数据处理。

这是一个使用多个进程计算数字平方的示例：

import multiprocessing


def compute_square(number):
    return number * number


if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]

    # create a pool of processes
    with multiprocessing.pool() as pool:
        # map function to numbers using multiple processes
        results = pool.map(compute_square, numbers)

    print("squares:", results)

# output:
# squares: [1, 4, 9, 16, 25]

以下是代码中的关键步骤：

函数compute_square被定义为接受一个数字作为输入并返回其平方。
if name == "main": 块中的代码确保它仅在直接执行脚本时运行。
定义了一个数字列表，将对其进行平方。
工作进程池是使用 multiprocessing.pool() 创建的。
map 方法用于将compute_square函数应用于列表中的每个数字，将工作负载分配到多个进程。

进程间通信（ipc）

由于每个进程都有自己的内存空间，进程之间共享数据需要进程间通信（ipc）机制。

多处理模块提供了一些用于ipc的工具，例如queue、pipe和value。

这是一个使用队列在进程之间共享数据的示例：

import multiprocessing


def worker(queue):
    # Retrieve and process data from the queue
    while not queue.empty():
        item = queue.get()
        print(f"Processing {item}")


if __name__ == "__main__":
    queue = multiprocessing.Queue()

    # Add items to the queue
    for i in range(10):
        queue.put(i)

    # Create a pool of processes to process the queue
    processes = []
    for _ in range(4):
        process = multiprocessing.Process(target=worker, args=(queue,))
        processes.append(process)
        process.start()

    # Wait for all processes to complete
    for process in processes:
        process.join()

    print("All processes have finished.")


# Output:
# Processing 0
# Processing 1
# Processing 2
# Processing 3
# Processing 4
# Processing 5
# Processing 6
# Processing 7
# Processing 8
# Processing 9
# All processes have finished.

在此示例中：

defworker(queue)：定义一个以队列作为参数的函数worker。该函数检索并处理队列中的项目，直到队列为空。
if name == "main":: 确保以下代码仅在直接执行脚本时运行，而不是作为模块导入时运行。
queue = multiprocessing.queue()：创建用于进程间通信的队列对象。
for i in range(10):queue.put(i)：将项目（数字 0 到 9）添加到队列中。
processes = []：初始化一个空列表来存储进程对象。
for _ in range(4) 循环：创建四个工作进程。
process = multiprocessing.process(target=worker, args=(queue,))：创建一个以worker为目标函数的新进程，并将队列作为参数传递。
processes.append(process)：将进程对象添加到进程列表中。
process.start()：启动进程。
进程中进程的 for 循环：使用 join() 方法等待每个进程完成。