如何基于指定字符串终止Python异步循环？

要实现动态迭代终止，核心是不要一次性生成所有任务，而是边执行边检测终止信号，同时控制并发数。下面是改造后的代码，解决手动设置count的问题，并且能在收到end标识时安全终止循环：

改造后的完整代码 #

import asyncio
import time
from httpx import AsyncClient, HTTPStatus

# 替换为你的实际配置变量
script = "your_script_id"
deploy = "your_deploy_id"
recordType = "your_record_type"
date = "your_date"
authHeaderSetup = lambda num: f"your_auth_header_{num}"  # 替换为真实的认证生成函数
limit = asyncio.Semaphore(4)  # 控制最大并发数为4
client = AsyncClient()

async def make_one_request(url: str, num: int):
    headers = {
        'Authorization': authHeaderSetup(num),
        'Content-Type': "application/json",
        'cache-control': "no-cache",
    }

    params = f'?script={script}&deploy={deploy}&record_type={recordType}&index={num}&date={date}'

    async with limit:
        print(f"发起请求 {num}")
        r = await client.get(url + params, headers=headers)

        if r.status_code != HTTPStatus.OK:
            raise ValueError(f"异常状态码: {r.status_code}")
        
        response_data = r.json()
        return num, response_data  # 返回索引和数据，方便跟踪请求状态

async def make_many_requests(url: str) -> list:
    results = []
    current_index = 0
    tasks = set()
    stop_flag = False

    while not stop_flag or tasks:
        # 未收到终止信号且未达并发上限时，启动新任务
        if not stop_flag and len(tasks) < limit._value:
            task = asyncio.create_task(make_one_request(url, current_index))
            tasks.add(task)
            current_index += 1
        
        # 等待任一任务完成
        done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        
        for task in done:
            tasks.remove(task)
            idx, data = await task
            
            if data == 'end':
                # 收到终止标识，停止创建新任务
                stop_flag = True
                print(f"在索引 {idx} 收到'end'，停止发起新请求...")
            else:
                results.append(data)
                print(f"请求 {idx} 完成")
    
    return results

if __name__ == "__main__":
    start = time.time()
    url = "your_restlet_base_url"
    final_results = asyncio.run(make_many_requests(url))
    
    print("\n最终结果:")
    print("==============")
    for res in final_results:
        print(res)
    
    end = time.time()
    print(f'总耗时: {end - start} 秒')
    asyncio.run(client.aclose())  # 关闭HTTP客户端连接，释放资源

关键改动说明 #

• 动态任务生成：不再预先创建所有任务，循环中根据并发上限动态启动新请求，直到收到终止信号。
• 安全终止逻辑：收到end标识时仅设置停止标志，不再启动新任务，但会等待所有已发起的请求完成，确保数据不丢失。
• 结果异步收集：用asyncio.wait监听任务完成事件，逐个处理返回结果，区分正常数据和终止信号。
• 代码优化：用f-string简化参数拼接，返回索引方便跟踪请求状态，最后主动关闭HTTP客户端避免资源泄漏。

注意事项 #

• 确保authHeaderSetup函数能为每个索引生成有效的认证头。
• 根据NetSuite的API频率限制调整并发数（当前设为4），避免触发限流。
• 如果NetSuite返回的end不是单独字符串，而是嵌套在响应结构中，需调整data == 'end'的判断逻辑。

内容的提问来源于stack exchange，提问作者Ryan Lebato