Python系列之Python并发执行实现方法

1、Python中并发执行实现方法
1.1 Python中并发执行实现

在Python中,有几种主要的并发执行实现方法,包括多线程、多进程和异步编程。

1.1.1 多线程(Threading)

Python标准库中的threading模块支持多线程编程。然而,由于Python的全局解释器锁(GIL),Python的多线程在CPU密集型任务上并不能实现真正的并行执行。但在I/O密集型任务(如网络请求、文件读写等)上,多线程仍然可以显著提升性能。

import threading  
  
def worker():  
    print("This is a thread running the worker function.")  
  
# 创建线程对象  
threads = []  
for _ in range(5):  
    t = threading.Thread(target=worker)  
    threads.append(t)  
    t.start()  
  
# 等待所有线程完成  
for t in threads:  
    t.join()
1.1.2 多进程(Multiprocessing)

Python的multiprocessing模块支持多进程编程,可以充分利用多核CPU的资源。每个进程都有自己独立的Python解释器,因此不受GIL的限制。多进程适用于CPU密集型任务。

import multiprocessing  
  
def worker():  
    print("This is a process running the worker function.")  
  
if __name__ == "__main__":  
    processes = []  
    for _ in range(5):  
        p = multiprocessing.Process(target=worker)  
        processes.append(p)  
        p.start()  
  
    # 等待所有进程完成  
    for p in processes:  
        p.join()
1.1.3 异步编程(Asyncio)

Python 3.5引入了asyncio模块,支持异步编程。异步编程可以在单线程内实现非阻塞的I/O操作,提高程序的响应速度和吞吐量。它特别适用于处理大量的并发I/O操作,如网络请求。

import asyncio  
  
async def worker():  
    print("This is an async task running the worker function.")  
  
# 创建事件循环  
loop = asyncio.get_event_loop()  
tasks = []  
for _ in range(5):  
    task = loop.create_task(worker())  
    tasks.append(task)  
  
# 执行所有任务  
loop.run_until_complete(asyncio.wait(tasks))  
loop.close()

注意:异步编程与多线程和多进程编程有所不同,它更多的是一种编程模型,而不是简单地创建多个执行单元。异步编程需要理解并适应其特有的编程模式和概念,如协程、事件循环等。

1.2 Python中多进程和多线程的区别

在Python中,multiprocessing和threading模块都用于实现并发执行,但它们在底层机制、使用场景和性能特点上有显著的区别。

1.2.1 Multiprocessing多进程

Multiprocessing模块允许创建多个进程来执行Python代码。每个进程都有自己独立的内存空间和解释器实例,因此它们之间不共享全局变量(除非通过特定的机制,如multiprocessing.Manager或multiprocessing.Value、multiprocessing.Array等)。这使得multiprocessing非常适合于计算密集型任务,因为它可以充分利用多核CPU并行处理任务。

由于每个进程都有自己独立的Python解释器,进程间通信(IPC)通常比线程间通信(通过共享内存)要慢得多,并且需要显式的IPC机制,如管道(Pipe)、队列(Queue)、共享内存(SharedMemory)等。

1.2.2 Threading多线程

Threading模块允许创建多个线程来执行Python代码。线程共享同一个进程的内存空间,因此它们可以直接访问全局变量和大多数Python对象。这使得线程间通信相对简单,因为它们可以直接读写共享的内存。

然而,由于Python的全局解释器锁(GIL)的存在,同一时间内只有一个线程可以执行Python代码。这意味着即使是多线程,CPU密集型任务的执行速度也可能不会显著提高。因此,threading模块在Python中通常更适用于IO密集型任务,如网络请求、文件读写等,这些任务通常可以在一个线程等待IO操作完成时让另一个线程继续执行。

1.2.3 多进程和多线程特性对比
  • 资源共享:
    • 多线程:在多线程中,所有线程共享同一个进程的地址空间,这意味着它们可以访问相同的变量和内存区域。因此,多线程间的数据共享和通信相对简单,但也容易引发数据同步和一致性的问题,如竞态条件。
    • 多进程:每个进程都有自己独立的地址空间,这意味着它们无法直接共享数据。进程间的通信需要通过特殊的机制来实现,如管道、消息队列、共享内存或套接字等。虽然进程间通信相对复杂,但它避免了多线程中的数据同步问题。
    • 全局解释器锁(GIL):
      • 多线程:由于Python的全局解释器锁(GIL)的存在,Python的多线程在CPU密集型任务上并不能实现真正的并行执行。GIL确保任何时候只有一个线程在执行Python字节码。因此,对于计算密集型任务,多线程并不能带来性能提升。
      • 多进程:多进程不受GIL的限制,每个进程都有自己独立的Python解释器,因此可以充分利用多核CPU的资源,实现真正的并行执行。
      • 性能开销:
        • 多线程:线程创建和销毁的开销相对较小,因为线程共享进程的内存空间,无需复制数据。因此,对于需要频繁创建和销毁线程的应用,多线程可能是一个更好的选择。
        • 多进程:进程创建和销毁的开销相对较大,因为每个进程都需要独立的内存空间和系统资源。此外,进程间通信也需要额外的开销。因此,对于需要大量进程的应用,需要谨慎考虑性能问题。
        • 稳定性:
          • 多线程:由于线程共享数据,如果一个线程崩溃,可能会导致整个进程崩溃。
          • 多进程:每个进程都是独立的,一个进程的崩溃不会影响其他进程。因此,多进程在稳定性方面可能更有优势。
          • 适用场景:
            • 多线程:适用于I/O密集型任务,如网络请求、文件读写等。在这些场景下,线程大部分时间都在等待I/O操作完成,因此可以充分利用多线程的优势。
            • 多进程:适用于CPU密集型任务,如科学计算、图像处理等。在这些场景下,多进程可以充分利用多核CPU的资源,实现性能提升。

              总之,选择使用多线程还是多进程取决于具体的任务类型和性能需求。在Python中,对于I/O密集型任务,可以使用多线程或异步编程;对于CPU密集型任务,多进程可能是一个更好的选择。

              1.3 等待信号量实现并发控制

              信号量是一个计数器,用于控制同时访问某个特定资源或资源池的线程数量。信号量有一个值,表示可用的许可数。当线程想要访问资源时,它必须先获取一个许可;如果许可数大于0,则获取成功并减1;否则,线程将阻塞等待。

              import threading  
              sem = threading.Semaphore(3)  # 允许三个线程同时访问资源  
              def worker():  
                  sem.acquire()  # 获取许可  
                  try:  
                      # 访问或修改共享资源  
                      print("Thread is working with the shared resource.")  
                  finally:  
                      sem.release()  # 释放许可  
              # 创建并启动线程...
              
              1.3.1 基于等待信号量实现多进程并发

              在Python中,基于等待信号量(Semaphore)实现多进程并发通常涉及到multiprocessing模块中的Semaphore类。信号量用于控制对共享资源的访问,允许一定数量的进程同时访问该资源。当信号量的值大于0时,进程可以获得一个信号量许可来访问资源;当信号量的值为0时,进程将阻塞,直到有其他进程释放一个许可。

              下面是一个简单的例子,展示了如何使用multiprocessing.Semaphore来实现多进程并发访问共享资源:

              import multiprocessing  
              import time  
              import random  
                
              # 设置信号量的初始值,这里允许3个进程同时访问共享资源  
              semaphore = multiprocessing.Semaphore(3)  
                
              def worker_process(process_id, semaphore):  
                  # 尝试获取信号量许可  
                  semaphore.acquire()  
                  try:  
                      print(f"Process {process_id} acquired semaphore and is working.")  
                      # 模拟工作负载  
                      time.sleep(random.random())  
                      print(f"Process {process_id} finished working and releasing semaphore.")  
                  finally:  
                      # 无论是否发生异常,都要确保释放信号量许可  
                      semaphore.release()  
                
              if __name__ == '__main__':  
                  # 创建进程池  
                  processes = []  
                  for i in range(10):  # 创建10个进程  
                      p = multiprocessing.Process(target=worker_process, args=(i, semaphore))  
                      processes.append(p)  
                      p.start()  
                
                  # 等待所有进程完成  
                  for p in processes:  
                      p.join()  
                
                  print("All processes have finished.")
              

              在这个例子中创建了10个进程,但是通过信号量限制了同时访问共享资源的进程数最多为3个。每个进程在工作前都会尝试获取信号量的许可,如果信号量的值大于0,则获取许可并开始工作;如果信号量的值为0,则进程会阻塞等待,直到有其他进程释放许可。每个进程完成工作后会释放其持有的信号量许可,这样其他等待的进程就可以获取许可并开始工作。

              1.3.2 基于等待信号量实现多线程并发

              在Python中,要实现基于等待信号量的多线程并发,可以使用threading模块中的Semaphore类。信号量用于控制对共享资源的并发访问。当信号量的值大于0时,线程可以获得一个信号量许可来访问资源;当信号量的值为0时,线程将阻塞,直到其他线程释放一个许可。

              下面是一个简单的例子,展示了如何使用threading.Semaphore来实现多线程并发访问共享资源:

              import threading  
              import time  
              import random  
                
              # 设置信号量的初始值,这里允许3个线程同时访问共享资源  
              semaphore = threading.Semaphore(3)  
                
              def worker_thread(thread_id, semaphore):  
                  # 尝试获取信号量许可  
                  semaphore.acquire()  
                  try:  
                      print(f"Thread {thread_id} acquired semaphore and is working.")  
                      # 模拟工作负载  
                      time.sleep(random.random())  
                      print(f"Thread {thread_id} finished working and releasing semaphore.")  
                  finally:  
                      # 无论是否发生异常,都要确保释放信号量许可  
                      semaphore.release()  
                
              if __name__ == '__main__':  
                  # 创建线程列表  
                  threads = []  
                  for i in range(10):  # 创建10个线程  
                      t = threading.Thread(target=worker_thread, args=(i, semaphore))  
                      threads.append(t)  
                      t.start()  
                
                  # 等待所有线程完成  
                  for t in threads:  
                      t.join()  
                
                  print("All threads have finished.")
              

              在这个例子中创建了10个线程,但是通过信号量限制了同时访问共享资源的线程数最多为3个。每个线程在工作前都会尝试获取信号量的许可,如果信号量的值大于0,则获取许可并开始工作;如果信号量的值为0,则线程会阻塞等待,直到有其他线程释放许可。每个线程完成工作后会释放其持有的信号量许可,这样其他等待的线程就可以获取许可并开始工作。