如何高效遍历超5万个GCP项目及近2万个存储桶？

现有代码问题分析 #

• 串行遍历完全依赖单线程处理所有IO请求，在5万+项目的场景下效率极低
• 当前多线程版本线程数固定为25，未结合GCP API配额做最优调整，且缺少分页优化、重试机制等关键提速手段

核心优化方案 #

1. 合理调整并发数 #

GCP Cloud Storage的list_buckets和getIamPolicy API有默认配额，建议根据实际配额调整线程/协程数：

• 一般可将线程池最大工作数设为50-100（需监控GCP控制台的配额使用情况，避免触发429限流）
• 若使用异步客户端，协程数可进一步提升至200-300，IO密集型场景下异步比线程池资源利用率更高

2. 分页参数优化 #

调用list_buckets时指定page_size=1000，单次API请求获取最大数量的存储桶，减少请求次数：

buckets = client.list_buckets(project=project, page_size=1000)

3. 添加重试机制 #

针对GCP API常见的限流（429）、临时错误（5xx），添加自动重试逻辑，避免请求失败中断流程。可使用GCP客户端自带的重试配置：

from google.api_core.retry import Retry

# 配置重试策略：针对429、500-599错误重试，最多5次
retry_policy = Retry(
    predicate=lambda exc: isinstance(exc, (google.api_core.exceptions.TooManyRequests, google.api_core.exceptions.ServerError)),
    maximum=5,
    multiplier=1.5,
)

# 调用API时传入重试参数
policy = bucket.get_iam_policy(retry=retry_policy)

4. 异步客户端改造 #

使用google-cloud-storage的异步客户端（需安装aiohttp），在IO密集型场景下比线程池更高效：

import asyncio
from google.cloud import storage_async

async def process_project(project):
    project_id = project['projectId']
    async_client = storage_async.Client(project=project_id)
    results = []
    try:
        buckets = async_client.list_buckets(page_size=1000)
        async for bucket in buckets:
            policy = await bucket.get_iam_policy()
            # 处理policy逻辑
            results.append((project_id, bucket.name))
    except Exception as e:
        print(f"处理项目{project_id}失败: {str(e)}")
    return results

async def main():
    # 获取项目列表逻辑
    projects = ...  # 此处保留原项目获取代码
    tasks = [process_project(p) for p in projects]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # 处理结果
    for res in results:
        if not isinstance(res, Exception):
            print(res)

if __name__ == "__main__":
    asyncio.run(main())

5. 减少同步IO操作 #

• 移除循环内的print语句，改为批量写入文件或日志系统，避免频繁IO阻塞
• 若不需要实时打印，可将结果缓存到内存后批量输出

优化后的多线程版本示例 #

import logging
from google.cloud import storage
from google.api_core.retry import Retry
from concurrent.futures import ThreadPoolExecutor

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 配置重试策略
retry_policy = Retry(
    predicate=lambda exc: isinstance(exc, (google.api_core.exceptions.TooManyRequests, google.api_core.exceptions.ServerError)),
    maximum=5,
    multiplier=1.5,
)

def get_bucket_policies(project):
    project_id = project['projectId']
    client = storage.Client(project=project_id)
    results = []
    try:
        # 分页优化
        buckets = client.list_buckets(project=project_id, page_size=1000)
        for bucket in buckets:
            # 带重试的IAM策略获取
            policy = bucket.get_iam_policy(retry=retry_policy)
            # 处理policy逻辑，这里保留原示例逻辑
            for test in policy:
                if 'name' in test:
                    pass  # 替换为实际业务逻辑
            results.append((project_id, bucket.name))
    except Exception as e:
        logger.error(f"项目{project_id}处理失败: {str(e)}")
    return results

def main():
    # 获取项目列表
    try:
        from googleapiclient.discovery import build
        from google.auth import default, DefaultCredentialsError
        from google.oauth2.credentials import Credentials
        projects = build('cloudresourcemanager', 'v1', credentials=Credentials.from_authorized_user_info(info=default()[1]), cache_discovery=False).projects().list().execute().get('projects', [])
    except DefaultCredentialsError:
        logger.exception("获取默认凭证失败")
        return

    # 调整线程数至50，根据配额可灵活调整
    with ThreadPoolExecutor(max_workers=50) as executor:
        futures = [executor.submit(get_bucket_policies, project) for project in projects]
        completed = 0
        total = len(projects)
        results = []
        for future in futures:
            try:
                res = future.result()
                results.extend(res)
            except Exception as e:
                logger.error(f"任务执行失败: {str(e)}")
            completed += 1
            logger.info(f"已完成 {completed}/{total} 个项目")

    # 批量输出结果
    with open('bucket_results.txt', 'w') as f:
        for project_id, bucket_name in results:
            f.write(f"{project_id} - {bucket_name}\n")

if __name__ == "__main__":
    main()

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