Python并发安全案例如何保障并发安全

本文目录导读：

1. 线程安全 - threading模块
2. 使用条件变量（Condition）
3. 使用Queue实现线程安全
4. 使用Event实现线程协调
5. 使用Semaphore控制并发数量
6. 进程安全 - multiprocessing模块
7. 使用全局解释器锁（GIL）特性
8. 使用futures模块
9. 原子操作示例
10. 并发安全的最佳实践

我来详细介绍Python中保障并发安全的常用方法和最佳实践。

线程安全 - threading模块

使用Lock（锁）

import threading
import time
class BankAccount:
    def __init__(self):
        self.balance = 0
        self.lock = threading.Lock()  # 创建锁
    def deposit(self, amount):
        with self.lock:  # 自动获取和释放锁
            current = self.balance
            time.sleep(0.1)  # 模拟IO操作
            self.balance = current + amount
            print(f"{threading.current_thread().name}: +${amount} -> {self.balance}")
    def withdraw(self, amount):
        with self.lock:
            if self.balance >= amount:
                current = self.balance
                time.sleep(0.1)
                self.balance = current - amount
                print(f"{threading.current_thread().name}: -${amount} -> {self.balance}")
            else:
                print(f"{threading.current_thread().name}: 余额不足")
# 测试
account = BankAccount()
threads = []
for i in range(5):
    t1 = threading.Thread(target=account.deposit, args=(100,), name=f"存款线程-{i}")
    t2 = threading.Thread(target=account.withdraw, args=(50,), name=f"取款线程-{i}")
    threads.extend([t1, t2])
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"最终余额: ${account.balance}")

使用RLock（可重入锁）

class Counter:
    def __init__(self):
        self.count = 0
        self.lock = threading.RLock()  # 可重入锁
    def increment(self):
        with self.lock:
            self.count += 1
            # 可以再次获取同一个锁（可重入）
            self.do_something_else()
    def do_something_else(self):
        with self.lock:  # 不会死锁
            pass
    def get_count(self):
        with self.lock:
            return self.count

使用条件变量（Condition）

import threading
class BoundedBuffer:
    def __init__(self, max_size=5):
        self.buffer = []
        self.max_size = max_size
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)
        self.not_empty = threading.Condition(self.lock)
    def put(self, item):
        with self.not_full:
            while len(self.buffer) >= self.max_size:
                print(f"缓冲区已满，生产者{threading.current_thread().name}等待...")
                self.not_full.wait()
            self.buffer.append(item)
            print(f"生产者{threading.current_thread().name}放入{item}, 缓冲区大小:{len(self.buffer)}")
            self.not_empty.notify()  # 通知消费者
    def get(self):
        with self.not_empty:
            while len(self.buffer) == 0:
                print(f"缓冲区为空，消费者{threading.current_thread().name}等待...")
                self.not_empty.wait()
            item = self.buffer.pop(0)
            print(f"消费者{threading.current_thread().name}取出{item}, 缓冲区大小:{len(self.buffer)}")
            self.not_full.notify()  # 通知生产者
            return item
# 测试
buffer = BoundedBuffer()
def producer():
    for i in range(10):
        buffer.put(f"产品-{i}")
def consumer():
    for _ in range(10):
        buffer.get()
threads = []
for i in range(2):
    t1 = threading.Thread(target=producer, name=f"P-{i}")
    t2 = threading.Thread(target=consumer, name=f"C-{i}")
    threads.extend([t1, t2])
for t in threads:
    t.start()
for t in threads:
    t.join()

使用Queue实现线程安全

from queue import Queue
import queue  # 注意导入的是queue模块
def worker_task(task_queue, result_queue, worker_id):
    while True:
        try:
            # 非阻塞方式获取任务
            task = task_queue.get_nowait()
            # 处理任务
            result = task * 2
            result_queue.put(result)
            print(f"工作者{worker_id}处理: {task} -> {result}")
            task_queue.task_done()
        except queue.Empty:
            break
# 创建队列
task_queue = Queue()
result_queue = Queue()
# 添加任务
for i in range(20):
    task_queue.put(i)
# 创建工作者线程
workers = []
for i in range(4):
    w = threading.Thread(target=worker_task, args=(task_queue, result_queue, i))
    workers.append(w)
    w.start()
# 等待所有任务完成
task_queue.join()
# 收集结果
results = []
while not result_queue.empty():
    results.append(result_queue.get())
print(f"所有结果: {sorted(results)}")

使用Event实现线程协调

import threading
import time
class DataProcessor:
    def __init__(self):
        self.data = None
        self.data_ready = threading.Event()
        self.data_processed = threading.Event()
    def producer(self):
        print("生产者: 准备数据...")
        time.sleep(2)
        self.data = {"name": "Python并发", "value": 100}
        print("生产者: 数据准备完毕")
        self.data_ready.set()  # 通知消费者
        self.data_processed.wait()  # 等待消费者处理完成
        print("生产者: 确认消费者已处理")
    def consumer(self):
        print("消费者: 等待数据...")
        self.data_ready.wait()  # 等待生产者
        print(f"消费者: 处理数据 {self.data}")
        time.sleep(1)
        self.data["processed"] = True
        print("消费者: 处理完成")
        self.data_processed.set()  # 通知生产者
# 测试
processor = DataProcessor()
t1 = threading.Thread(target=processor.producer)
t2 = threading.Thread(target=processor.consumer)
t1.start()
t2.start()
t1.join()
t2.join()

使用Semaphore控制并发数量

import threading
import time
class WebScraper:
    def __init__(self, max_concurrent=3):
        self.semaphore = threading.Semaphore(max_concurrent)
    def scrape_url(self, url, scraper_id):
        with self.semaphore:
            print(f"爬虫{scraper_id}: 开始爬取 {url}")
            time.sleep(2)  # 模拟网络请求
            print(f"爬虫{scraper_id}: 完成爬取 {url}")
# 测试
scraper = WebScraper(max_concurrent=3)
urls = [f"http://example.com/page/{i}" for i in range(10)]
threads = []
for i, url in enumerate(urls):
    t = threading.Thread(target=scraper.scrape_url, args=(url, i))
    threads.append(t)
    t.start()
for t in threads:
    t.join()

进程安全 - multiprocessing模块

from multiprocessing import Process, Value, Array, Lock
import time
class SharedCounter:
    def __init__(self):
        self.count = Value('i', 0)  # 共享整数
        self.lock = Lock()
    def increment(self, amount):
        with self.lock:
            self.count.value += amount
            print(f"进程{threading.current_thread().name}: 计数 {self.count.value}")
            time.sleep(0.1)
def worker(counter, amount):
    for _ in range(10):
        counter.increment(amount)
# 测试
counter = SharedCounter()
processes = []
for i in range(3):
    p = Process(target=worker, args=(counter, 1))
    processes.append(p)
    p.start()
for p in processes:
    p.join()
print(f"最终计数: {counter.count.value}")

使用全局解释器锁（GIL）特性

import threading
import time
# CPU密集型任务 - 受GIL影响
class CPUTask:
    def compute(self):
        total = 0
        for _ in range(50000000):
            total += 1
        return total
# 使用多进程处理CPU密集型任务
def cpu_intensive_task(name):
    total = 0
    for i in range(10000000):
        total += i
    return f"{name}: {total}"
# 最佳实践：根据不同场景选择并发模型
def process_by_type(task_type, data):
    if task_type == "IO":
        # IO密集型任务使用多线程
        threads = []
        for item in data:
            t = threading.Thread(target=handle_io, args=(item,))
            threads.append(t)
            t.start()
        for t in threads:
            t.join()
    else:
        # CPU密集型任务使用多进程
        from concurrent.futures import ProcessPoolExecutor
        with ProcessPoolExecutor() as executor:
            results = executor.map(handle_cpu, data)

使用futures模块

from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
def thread_safe_operation(value, lock):
    with lock:
        print(f"处理值: {value}")
        return value * 2
lock = threading.Lock()
values = list(range(20))
# 使用线程池
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(thread_safe_operation, v, lock): v for v in values}
    for future in as_completed(futures):
        result = future.result()
        print(f"结果: {result}")

原子操作示例

import threading
class AtomicCounter:
    def __init__(self):
        self._lock = threading.Lock()
        self._count = 0
    def increment(self):
        # 使用原子操作（通过锁实现）
        with self._lock:
            self._count += 1
    @property
    def count(self):
        with self._lock:
            return self._count
# 虽然Python有GIL，但多线程下仍需注意
class NonSafeCounter:
    def __init__(self):
        self.count = 0
    def increment(self):
        # 非原子操作，可能出问题
        count = self.count
        count += 1
        self.count = count  # 这里可能被其他线程打断

并发安全的最佳实践

import threading
from contextlib import contextmanager
class ConnectionPool:
    def __init__(self, max_connections=5):
        self._connections = []
        self._max = max_connections
        self._lock = threading.Lock()
        self._cond = threading.Condition(self._lock)
    @contextmanager
    def get_connection(self):
        with self._cond:
            while len(self._connections) >= self._max:
                self._cond.wait()
            conn = self._create_connection()
            self._connections.append(conn)
        try:
            yield conn
        finally:
            with self._cond:
                self._connections.remove(conn)
                self._cond.notify()
    def _create_connection(self):
        return "Connection-" + str(threading.current_thread().ident)
# 使用示例
pool = ConnectionPool(max_connections=3)
def use_connection(thread_id):
    with pool.get_connection() as conn:
        print(f"线程{thread_id}: 使用连接 {conn}")
        import time
        time.sleep(0.5)
threads = []
for i in range(10):
    t = threading.Thread(target=use_connection, args=(i,))
    threads.append(t)
    t.start()
for t in threads:
    t.join()

1. 选择正确的锁类型：

   • Lock: 基本互斥锁
   • RLock: 可重入锁，适合递归调用
   • Semaphore: 控制并发数量
   • Event: 线程间信号通知
   • Condition: 条件变量，复杂协调

2. 安全准则：

   • 尽量使用 with 语句自动管理锁
   • 避免死锁（注意锁的顺序）
   • 最小化锁的持有时间
   • 考虑使用 queue.Queue 代替自定义同步

3. 性能考虑：

   • IO密集型任务使用多线程
   • CPU密集型任务使用多进程
   • 使用线程池/进程池管理资源

4. 调试技巧：

   • 使用 logging 而非 print 避免输出混合
   • 添加适当的错误处理
   • 使用 timeout 参数避免无限等待

这些示例涵盖了Python并发安全的主要场景和最佳实践,根据具体需求选择合适的同步机制是关键。