Python脚本如何精准控制同步执行范围

在Python中精准控制同步执行范围,主要有以下几种方法：

使用 with 语句和锁（最常用）

线程锁 threading.Lock

import threading
lock = threading.Lock()
shared_data = []
def safe_append(item):
    with lock:  # 只有在这个代码块内是同步的
        shared_data.append(item)
        # 其他同步操作...
    # 这里之后不同步

递归锁 threading.RLock

rlock = threading.RLock()
def nested_function():
    with rlock:
        # 可以再次获取锁
        with rlock:
            pass

使用信号量控制并发数量

import threading
import time
semaphore = threading.Semaphore(3)  # 最多3个线程同时执行
def limited_worker():
    with semaphore:  # 只有在这个范围内控制并发数
        print(f"线程 {threading.current_thread().name} 开始工作")
        time.sleep(2)
        print("工作完成")
    # 这里之后不受限制

使用条件变量精细控制

import threading
condition = threading.Condition()
buffer = []
BUFFER_SIZE = 5
def producer():
    with condition:
        while len(buffer) >= BUFFER_SIZE:
            condition.wait()  # 等待消费者通知
        buffer.append(item)
        condition.notify()  # 通知消费者
def consumer():
    with condition:
        while not buffer:
            condition.wait()
        item = buffer.pop(0)
        condition.notify()

使用事件对象控制执行时机

import threading
event = threading.Event()
def waiter():
    event.wait()  # 等待事件触发
    # 执行同步的代码块
    print("事件已触发，开始执行")
def setter():
    time.sleep(5)
    event.set()  # 触发事件

使用 contextlib.contextmanager 自定义同步范围

from contextlib import contextmanager
import threading
@contextmanager
def synchronized(lock):
    lock.acquire()
    try:
        yield  # 这里是同步执行的代码块
    finally:
        lock.release()
# 使用
lock = threading.Lock()
with synchronized(lock):
    # 这些代码是同步执行的
    shared_resource.modify()

使用队列控制任务同步

from queue import Queue
import threading
task_queue = Queue(maxsize=10)
def worker():
    while True:
        task = task_queue.get()  # 获取任务，如果没有则阻塞
        # 处理任务
        task_queue.task_done()
# 生产者
def producer():
    for i in range(100):
        task_queue.put(i)  # 放入任务，如果队列满了则阻塞

精确的时间控制

import threading
import time
class PreciseScheduler:
    def __init__(self):
        self.ready_time = time.time()
        self.lock = threading.Lock()
    def wait_until(self, target_time):
        with self.lock:
            wait_time = target_time - time.time()
            if wait_time > 0:
                time.sleep(wait_time)
            # 精确到这里的代码在目标时间执行
            self.execute_target_code()

使用 threading.Barrier 同步多个线程

import threading
barrier = threading.Barrier(5)  # 等待5个线程
def thread_work():
    # 准备阶段...
    barrier.wait()  # 等待所有线程到达此处
    # 所有线程同时开始执行这里的代码

最佳实践建议

1. 最小化同步范围：

   # 不好的做法
   with lock:
    do_something_long()  # 长时间占用锁
    do_critical_work()   # 只需要保护这个
    do_another_slow()    # 不必要的同步

好的做法

do_something_long() with lock: do_critical_work() # 只保护必要的部分 do_another_slow()

2. **使用 `try...finally` 确保释放**：
```python
lock.acquire()
try:
    # 同步代码
finally:
    lock.release()

3. 避免死锁：

   # 使用with语句自动管理
   with lock1:
    with lock2:
        pass

通过这些方法，你可以精确控制Python中需要同步执行的代码范围,确保线程安全的同时最大化并发性能。