并发编程中的锁机制：简单和复杂同步操作示例

原创川长思鸟来 2024-12-15 00:48 20阅读 0赞

在并发编程中，锁机制是一种资源管理和互斥访问的手段。简单和复杂的同步操作可以用来控制多个线程对共享数据的访问。

1. **简单锁机制（如Python的`threading.Lock`）**：

- 示例：只有一个生产者和一个消费者。
   
   ```python
   import threading

# 创建一个锁对象
   lock = threading.Lock()

def producer():
       for _ in range(10):
           with lock:
               print("Produce: Item", _)
           time.sleep(0.2)  # 模拟生产耗时

def consumer():
       while True:
           with lock:
               if not produce_thread.is_alive():  # 判断生产线程是否存活
                   print("Consume: Wait for item")
                   continue  # 等待下一次循环
               item = lock.acquire()  # 获取锁，完成资源的保护
               print("Consume: Item", item)
               lock.release(item)  # 释放锁，保证其他线程能访问该资源

produce_thread = threading.Thread(target=producer))
   consumer_thread = threading.Thread(target=consumer))

produce_thread.start()
   consumer_thread.start()

producer_thread.join()  # 等待生产线程结束
   consumer_thread.join()  # 等待消费者线程结束
   ```

2. **复杂锁机制（如`concurrent.futures`中的锁或数据库中的乐观锁/悲观锁）**：

- 示例：多个生产者和一个消费者，采用分布式锁实现。
   
   ```python
   import concurrent.futures

class DistributedLock:
       def __init__(self, lock_key):
           self.lock_key = lock_key
           self.locked_by = None

def acquire(self):
           if not self.is_locked():
               # 使用分布式哈希算法确定锁的存储位置
               lock_store = self.get_lock_store()
               lock_key = f"{self.lock_key}:#{lock_store}::{self.__class__.__name__}"
               lock = Lock(lock_key)
               lock.locked_by = self
               return lock  # 返回锁定对象

def is_locked(self):
           if self.locked_by:
               return True
           else:
               lock_stores = [x[1] for x in self.get_locks()]  # 获取所有锁的存储位置列表
               lock_keys = [f"{self.lock_key}:#{store}::{self.__class__.__name__}" for store in lock_stores]
               for lock_key in lock_keys:
                   if self.check_lock(lock_key)):
                       return True
           return False

def get_locks(self):
           # 根据锁的存储位置，获取对应的锁对象列表
           pass

def check_lock(self, lock_key):
           # 检查给定的锁键是否已被锁定
           pass

def get_lock_store(self):
           # 获取当前锁的存储位置
           pass

class DistributedProducer:
       def __init__(self, lock_key, num_items):
           self.lock_key = lock_key
           self.num_items = num_items
           self.item_count = 0
           self.producer_thread = threading.Thread(target=self.produce)

def produce(self):
           with self.get_lock() as lock:
               if not lock.is_locked_by(self):  # 检查当前锁是否属于本生产者
                   print(f"{self.lock_key}: 已被其他生产者锁定")
                   time.sleep(2)  # 等待解锁，避免竞争和死锁
               self.item_count += 1
               if self.item_count >= self.num_items:  # 当生产完成指定数量时，自动停止生产
                   print(f"{self.lock_key}: 生产完成，停止生产和锁定资源"))
                   lock.unlock(lock.is_locked_by(self))  # 解锁并释放锁资源
                   break
               else:
                   print(f"{self.lock_key}: 开始生成第{self.item_count}个物品..."))
                   time.sleep(0.5)  # 模拟生产耗时

def get_lock(self):
           return DistributedLock(lock_key=self.lock_key))

class Consumer:
       def __init__(self, lock_key):
           self.lock_key = lock_key
           self.consumer_thread = threading.Thread(target=self.consume)

def consume(self):
           with self.get_lock() as lock:
               if not lock.is_locked_by(self):  # 检查当前锁是否属于本消费者
                   print(f"{self.lock_key}: 已被其他消费者锁定"))
                   time.sleep(2)  # 等待解锁，避免竞争和死锁
               else:
                   print(f"{self.lock_key}: 开始消费第{lock.item_count}个物品..."))
                   lock.item_count += 1  # 模拟消费操作，增加计数器的值
                   time.sleep(0.5)  # 模拟消费耗时

def get_lock(self):
           return DistributedLock(lock_key=self.lock_key))

producer = DistributedProducer(lock_key="producer_key", num_items=10)
   consumer = Consumer(lock_key="consumer_key"))

producer.produce()
   consumer.consume()

producer.producer_thread.join()  # 等待生产者线程结束
   consumer.consumer_thread.join()  # 等待消费者线程结束

```

这是一段使用分布式锁和多线程实现的并发编程示例。在这个例子中，`DistributedLock`被用来管理锁资源，确保在多生产者、多消费者的场景下，对共享资源进行有效的互斥访问。