Google Drive Files:list API超大账号查询提速方案咨询

针对单账号处理超大文件量的瓶颈，给你几个实用的提速方案：

• 并发请求榨干单账号配额
  Google Drive API单账号默认有1000请求/100秒的配额，完全可以用线程池（比如concurrent.futures.ThreadPoolExecutor）同时发起多个分页请求，利用等待响应的间隙处理其他请求。注意控制并发数在5-10左右，别超过配额触发限流。

• 砍字段减少数据传输
  你当前请求的字段里如果有实际用不上的（比如parents、mimeType或者权限里的deleted），直接从fields参数里删掉。返回的数据量越小，传输和解析的速度越快，能省不少时间。

• 增量查询替代全量遍历
  如果是重复执行的任务，别每次都扫2800万文件。记录上次处理的时间戳，用q参数过滤出更新过的文件：q="'curr_user' in owners and modifiedTime > '2024-01-01T00:00:00Z'"，只处理新增或变更的文件。

• 批量请求减少连接开销
  用Google API的批量请求功能，把多个分页请求打包成一个HTTP请求发送，减少TCP连接建立的耗时。官方库的BatchHttpRequest类可以直接实现这个操作，一次性提交多个files().list请求。

• 本地处理异步化/批量化
  检查拿到文件数据后的处理逻辑，如果有写数据库、打日志这类IO操作，别每条数据都同步执行。攒够一批（比如1000条）再一次性写入，或者用异步IO处理，避免IO阻塞拖慢整体流程。

线程池改造的示例代码 #

from concurrent.futures import ThreadPoolExecutor, as_completed
import logging

def gw_list_files_page(service, curr_user, nextPageToken):
    try:
        if nextPageToken:
            results = service.files().list(
                pageToken=str(nextPageToken),
                pageSize=1000,
                q=f"'{curr_user}' in owners",
                fields="nextPageToken,incompleteSearch,files(id,name,shared,permissions(type,emailAddress,role))"
            ).execute()
        else:
            results = service.files().list(
                pageSize=1000,
                q=f"'{curr_user}' in owners",
                fields="nextPageToken,incompleteSearch,files(id,name,shared,permissions(type,emailAddress,role))"
            ).execute()
        return results
    except Exception as e:
        logging.error(f"{curr_user}: 列表文件出错 - {str(e)}")
        return None

def process_files_batch(files_batch):
    # 这里写你的文件权限处理逻辑，比如解析、入库等
    pass

def list_files_concurrently(service, curr_user, max_workers=5):
    items_count = 0
    nextPageTokens = [None]
    processed_tokens = set()

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        while nextPageTokens:
            futures = []
            # 提交未处理的分页请求
            for token in nextPageTokens:
                if token not in processed_tokens:
                    processed_tokens.add(token)
                    futures.append(executor.submit(gw_list_files_page, service, curr_user, token))
            
            # 处理返回结果
            for future in as_completed(futures):
                results = future.result()
                if not results:
                    continue
                
                files = results.get('files', [])
                items_count += len(files)
                logging.debug(f"{curr_user}: 已处理 {items_count} 个文件")
                process_files_batch(files)
                
                # 收集下一页token
                next_token = results.get('nextPageToken')
                if next_token and next_token not in processed_tokens:
                    nextPageTokens.append(next_token)
            
            # 清理已处理的token
            nextPageTokens = [t for t in nextPageTokens if t not in processed_tokens]

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