Python Semaphore信号量详解 - 多线程限流与资源池管理

import threading

# 创建信号量，指定允许同时访问的线程数
semaphore = threading.Semaphore(3)  # 允许3个线程同时访问

# 基本使用方式
semaphore.acquire()  # 获取信号量，计数器-1
try:
    # 访问共享资源
    pass
finally:
    semaphore.release()  # 释放信号量，计数器+1

# 推荐使用with语句
with semaphore:
    # 访问共享资源（自动管理）
    pass

import threading
import time

# 演示Semaphore计数器变化
sem = threading.Semaphore(2)
print(f"初始状态")

# 线程1获取
sem.acquire()
print("线程1获取信号量")

# 线程2获取
sem.acquire()
print("线程2获取信号量")

# 线程3尝试获取（会阻塞）
print("线程3尝试获取（将阻塞）...")

def worker3():
    sem.acquire()
    print("线程3获取成功！")
    sem.release()

t = threading.Thread(target=worker3)
t.start()

time.sleep(0.5)
print("释放一个信号量")
sem.release()  # 线程3会被唤醒

t.join()
sem.release()
print("完成")

import threading
import time
import random

# 模拟最多3个并发下载
download_semaphore = threading.Semaphore(3)

def download_file(file_id):
    with download_semaphore:
        print(f"[开始] 下载文件 {file_id}，当前线程: {threading.current_thread().name}")
        # 模拟下载耗时
        time.sleep(random.uniform(0.5, 2))
        print(f"[完成] 文件 {file_id} 下载完成")

# 创建10个下载任务
files = [f"file_{i}.zip" for i in range(10)]
threads = []

for file_id in files:
    t = threading.Thread(target=download_file, args=(file_id,))
    threads.append(t)
    t.start()

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

print("所有文件下载完成")

import threading
import time
import random

class ConnectionPool:
    def __init__(self, pool_size=3):
        self.pool_size = pool_size
        self.semaphore = threading.Semaphore(pool_size)
        self.connections = [f"conn_{i}" for i in range(pool_size)]
        self.lock = threading.Lock()
        self.available = self.connections.copy()
    
    def get_connection(self):
        self.semaphore.acquire()
        with self.lock:
            conn = self.available.pop()
            print(f"获取连接: {conn}, 剩余可用: {len(self.available)}")
            return conn
    
    def release_connection(self, conn):
        with self.lock:
            self.available.append(conn)
            print(f"释放连接: {conn}, 剩余可用: {len(self.available)}")
        self.semaphore.release()

# 使用连接池
pool = ConnectionPool(3)

def db_task(task_id):
    conn = pool.get_connection()
    try:
        print(f"任务{task_id}使用{conn}执行查询")
        time.sleep(random.uniform(0.5, 1.5))
    finally:
        pool.release_connection(conn)

# 创建多个任务
threads = [threading.Thread(target=db_task, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print("所有数据库任务完成")

import threading
import time
import random

# 限制同时最多5个API请求
api_semaphore = threading.Semaphore(5)

class APIClient:
    def __init__(self):
        self.request_count = 0
        self.lock = threading.Lock()
    
    def call_api(self, endpoint):
        with api_semaphore:
            with self.lock:
                self.request_count += 1
                count = self.request_count
            
            print(f"[请求{count}] 调用API: {endpoint}")
            # 模拟API调用
            time.sleep(random.uniform(0.3, 1))
            print(f"[响应{count}] API返回结果")

client = APIClient()

def request_worker(request_id):
    client.call_api(f"/api/data/{request_id}")

# 创建20个请求
threads = [threading.Thread(target=request_worker, args=(i,)) for i in range(20)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"总请求数: {client.request_count}")