Python多线程和多进程深度解析

作为一名从后端开发转向Rust的开发者，我发现Python的多线程和多进程与Rust的并发模型有很多相似之处，但也有一些不同。Python的多线程和多进程可以用于并发执行任务，提高程序的性能。今天我想分享一下我对Python多线程和多进程的理解和实践。

多线程的基本概念

线程是进程内的一个执行单元，多个线程可以共享进程的资源。Python的多线程通过threading模块实现。

import threading
import time

def worker():
    print(f"Worker thread {threading.current_thread().name} started")
    time.sleep(1)
    print(f"Worker thread {threading.current_thread().name} completed")

# 创建线程
threads = []
for i in range(5):
    t = threading.Thread(target=worker, name=f"Thread-{i}")
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("All threads completed")

多进程的基本概念

进程是操作系统分配资源的基本单位，每个进程有自己的内存空间。Python的多进程通过multiprocessing模块实现。

import multiprocessing
import time

def worker():
    print(f"Worker process {multiprocessing.current_process().name} started")
    time.sleep(1)
    print(f"Worker process {multiprocessing.current_process().name} completed")

# 创建进程
processes = []
for i in range(5):
    p = multiprocessing.Process(target=worker, name=f"Process-{i}")
    processes.append(p)
    p.start()

# 等待所有进程完成
for p in processes:
    p.join()

print("All processes completed")

线程安全

由于多个线程共享进程的资源，需要注意线程安全问题。Python提供了多种同步原语来解决线程安全问题。

1. 锁（Lock）

import threading
import time

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(1000000):
        with lock:
            counter += 1

# 创建线程
threads = []
for i in range(5):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print(f"Final counter value: {counter}")

2. 条件变量（Condition）

import threading
import time

condition = threading.Condition()
data = []

def producer():
    for i in range(5):
        with condition:
            data.append(i)
            print(f"Produced: {i}")
            condition.notify()  # 通知消费者
        time.sleep(0.5)

def consumer():
    for _ in range(5):
        with condition:
            while not data:
                condition.wait()  # 等待生产者通知
            item = data.pop(0)
            print(f"Consumed: {item}")

# 创建线程
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()

3. 信号量（Semaphore）

import threading
import time

# 最多允许3个线程同时访问
semaphore = threading.Semaphore(3)

def worker(i):
    with semaphore:
        print(f"Worker {i} started")
        time.sleep(1)
        print(f"Worker {i} completed")

# 创建线程
threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("All workers completed")

进程间通信

由于进程之间不共享内存，需要使用进程间通信（IPC）机制来交换数据。

1. 队列（Queue）

import multiprocessing
import time

def producer(queue):
    for i in range(5):
        queue.put(i)
        print(f"Produced: {i}")
        time.sleep(0.5)

def consumer(queue):
    for _ in range(5):
        item = queue.get()
        print(f"Consumed: {item}")
        time.sleep(1)

# 创建队列
queue = multiprocessing.Queue()

# 创建进程
producer_process = multiprocessing.Process(target=producer, args=(queue,))
consumer_process = multiprocessing.Process(target=consumer, args=(queue,))

producer_process.start()
consumer_process.start()

producer_process.join()
consumer_process.join()

2. 管道（Pipe）

import multiprocessing
import time

def sender(conn):
    for i in range(5):
        conn.send(i)
        print(f"Sent: {i}")
        time.sleep(0.5)
    conn.close()

def receiver(conn):
    while True:
        try:
            item = conn.recv()
            print(f"Received: {item}")
            time.sleep(1)
        except EOFError:
            break

# 创建管道
parent_conn, child_conn = multiprocessing.Pipe()

# 创建进程
sender_process = multiprocessing.Process(target=sender, args=(child_conn,))
receiver_process = multiprocessing.Process(target=receiver, args=(parent_conn,))

sender_process.start()
receiver_process.start()

sender_process.join()
receiver_process.join()

3. 共享内存（Shared Memory）

import multiprocessing
import time

def increment(counter):
    for _ in range(1000000):
        with counter.get_lock():
            counter.value += 1

# 创建共享内存
counter = multiprocessing.Value('i', 0)

# 创建进程
processes = []
for i in range(5):
    p = multiprocessing.Process(target=increment, args=(counter,))
    processes.append(p)
    p.start()

# 等待所有进程完成
for p in processes:
    p.join()

print(f"Final counter value: {counter.value}")

线程池和进程池

使用线程池和进程池可以更高效地管理线程和进程。

1. 线程池

import concurrent.futures
import time

def worker(i):
    print(f"Worker {i} started")
    time.sleep(1)
    print(f"Worker {i} completed")
    return i * 2

# 创建线程池
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务
    futures = [executor.submit(worker, i) for i in range(10)]
    
    # 收集结果
    results = [future.result() for future in concurrent.futures.as_completed(futures)]

print(f"Results: {results}")

2. 进程池

import concurrent.futures
import time

def worker(i):
    print(f"Worker {i} started")
    time.sleep(1)
    print(f"Worker {i} completed")
    return i * 2

# 创建进程池
with concurrent.futures.ProcessPoolExecutor(max_workers=3) as executor:
    # 提交任务
    futures = [executor.submit(worker, i) for i in range(10)]
    
    # 收集结果
    results = [future.result() for future in concurrent.futures.as_completed(futures)]

print(f"Results: {results}")

多线程与多进程的选择

适合使用多线程的场景

1. I/O密集型任务：如文件操作、网络请求等，因为线程在等待I/O操作时会释放GIL，允许其他线程执行。
2. 需要共享内存的任务：线程之间共享内存，数据交换更高效。
3. 启动速度快的任务：线程的启动速度比进程快。

适合使用多进程的场景

1. CPU密集型任务：如数学计算、图像处理等，因为多进程可以利用多核CPU，避免GIL的限制。
2. 需要隔离的任务：进程之间相互隔离，一个进程崩溃不会影响其他进程。
3. 需要大量内存的任务：每个进程有自己的内存空间，避免内存竞争。

多线程和多进程与Rust的对比

相似之处

• 都支持并发执行任务
• 都提供了同步原语来解决并发问题
• 都支持线程池和进程池
• 都可以处理I/O密集型和CPU密集型任务

不同之处

• Python的多线程受GIL限制，而Rust的线程没有此限制
• Python的多进程启动较慢，而Rust的线程启动较快
• Python的并发模型相对简单，而Rust的并发模型更加安全和强大
• Python的多线程和多进程使用不同的模块，而Rust的并发功能集成在标准库中

实战案例：使用多线程和多进程处理任务

import concurrent.futures
import time
import requests

# 模拟I/O密集型任务
def fetch_url(url):
    response = requests.get(url)
    return url, response.status_code

# 模拟CPU密集型任务
def compute_factorial(n):
    result = 1
    for i in range(1, n + 1):
        result *= i
    return n, result

def main():
    # 测试I/O密集型任务（使用线程池）
    urls = [
        "https://www.google.com",
        "https://www.github.com",
        "https://www.python.org",
        "https://www.rust-lang.org",
        "https://www.csdn.net"
    ]
    
    print("Testing I/O密集型任务 with ThreadPoolExecutor:")
    start_time = time.time()
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(fetch_url, urls))
    
    for url, status_code in results:
        print(f"{url}: {status_code}")
    
    print(f"I/O密集型任务 took {time.time() - start_time:.2f} seconds")
    
    # 测试CPU密集型任务（使用进程池）
    numbers = [100000, 200000, 300000, 400000, 500000]
    
    print("\nTesting CPU密集型任务 with ProcessPoolExecutor:")
    start_time = time.time()
    
    with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(compute_factorial, numbers))
    
    for n, result in results:
        print(f"{n}! has {len(str(result))} digits")
    
    print(f"CPU密集型任务 took {time.time() - start_time:.2f} seconds")

if __name__ == "__main__":
    main()

总结

Python的多线程和多进程是一种强大的工具，它们可以帮助我们并发执行任务，提高程序的性能。通过threading和multiprocessing模块，我们可以轻松地创建和管理线程