Python异步编程中asyncio.gather的并发控制详解

import asyncio
 
async def task(n):
    print(f"Task {n} started")
    await asyncio.sleep(1)
    print(f"Task {n} completed")
    return n
 
async def main():
    tasks = [task(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    print(f"Total results: {len(results)}")
 
asyncio.run(main())

async def controlled_task(sem, n):
    async with sem:  # 获取信号量
        print(f"Task {n} acquired semaphore")
        await asyncio.sleep(1)
        print(f"Task {n} released semaphore")
        return n
 
async def main():
    sem = asyncio.Semaphore(3)  # 最大并发3
    tasks = [controlled_task(sem, i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    print(f"Total results: {len(results)}")
 
asyncio.run(main())

async def dynamic_control():
    sem = asyncio.Semaphore(5)
    task_queue = asyncio.Queue()
    
    # 生产者
    async def producer():
        for i in range(20):
            await task_queue.put(i)
    
    # 消费者
    async def consumer():
        while True:
            item = await task_queue.get()
            async with sem:
                print(f"Processing {item}")
                await asyncio.sleep(1)
            task_queue.task_done()
    
    # 动态调整
    def monitor(queue):
        while True:
            size = queue.qsize()
            if size > 10:
                sem._value = max(1, sem._value - 1)
            elif size < 5:
                sem._value = min(10, sem._value + 1)
            asyncio.sleep(1)
    
    await asyncio.gather(
        producer(),
        *[consumer() for _ in range(3)],
        asyncio.to_thread(monitor, task_queue)
    )
 
asyncio.run(dynamic_control())

def chunked(iterable, chunk_size):
    for i in range(0, len(iterable), chunk_size):
        yield iterable[i:i+chunk_size]
 
async def batch_processing():
    all_tasks = [task(i) for i in range(100)]
    
    for batch in chunked(all_tasks, 10):
        print(f"Processing batch: {len(batch)} tasks")
        await asyncio.gather(*batch)
 
asyncio.run(batch_processing())

try:
    await asyncio.wait_for(task(), timeout=10)
except asyncio.TimeoutError:
    print("Task timed out")

# 错误示例：缺少async with
sem = asyncio.Semaphore(3)
sem.acquire()
await task()
sem.release()  # 容易忘记释放

# 正确用法
async with sem:
    await task()  # 自动获取和释放

async def risky_task():
    async with sem:
        raise Exception("Oops!")  # 异常导致sem未释放

async def safe_task():
    sem_acquired = False
    try:
        async with sem:
            sem_acquired = True
            # 执行可能出错的操作
    finally:
        if not sem_acquired:
            sem.release()