如何用Promise与async/await实现带分页的多属性并行数据拉取

我正在研究用Promise和async/await从带分页的REST API拉取数据，具体场景是通过MS Graph API获取Azure用户的完整数据——因为Azure不会单次返回所有用户数据，必须多次调用才能拼全。

现有用户基础数据分页逻辑 #

目前已经实现了用户基础数据的分页拉取逻辑，代码运行正常：

let nextPageLink: string| null = null;
do {
      let pagedResult: AggregationPageResult = await this.fetchNextPage(nextPageLink);
      let usersMap: Map<string, any>[] = pagedResult.data!;
      nextPageLink = pagedResult.nextPageLink!;
      this.sendPageResult(usersMap, res);
} while(nextPageLink) 

用户组信息拉取的问题 #

但现在获取用户组成员信息的代码只能处理单页数据，如果用户组数量超过API默认页大小（比如用户有1000个组，但默认只返回100个），就没法拉取完整数据——因为没处理这个接口的nextPageLink分页逻辑。当前组信息拉取代码如下：

for(let user of usersToSearch) {
    let groupMembershipAPI = "https://graph.microsoft.com/v1.0/users/" + user + "/memberOf/microsoft.graph.group";
     
     let httpConfig: HttpRequestConfig = {
             baseURL: groupMembershipAPI, 
             method: 'GET',
             queryFilter: filters
         }
     
      executions.push(httpClient.executeRequest(httpConfig, true));
}    

let executionResults: HttpResponse [] = await Promise.all(executions);

需求目标 #

除此之外，我还需要拉取用户的目录角色、PIM数据、最后登录活动等更多属性，同时希望实现以下目标：

• 按顺序同步处理用户的分页数据
• 并行拉取用户的各类属性数据（比如组和角色拉取无需互相等待）
• 找到更优的Promise链式/动态处理方案，兼顾性能和资源占用

用伪代码描述需求如下：

async public fetchNextPage(nextPageLink: string): AggregationPageResult {
    let propertyFetchResults = []
     1. await fetch the next user page
     2. for every extra property to be fetched with pagination
          2.a propertyFetchResults.push (...result)
     3. await Promise.all(propertyFetchResults)
     4. merge the above property results with users of current page
    
 }

解决方案实现 #

1. 通用分页拉取函数 #

先封装一个通用的分页拉取函数，用来处理任何需要分页的MS Graph接口：

async function fetchAllPages<T>(initialUrl: string, httpClient: any, filters?: any): Promise<T[]> {
    let allItems: T[] = [];
    let nextPageLink: string | null = initialUrl;

    while (nextPageLink) {
        const config: HttpRequestConfig = {
            baseURL: nextPageLink,
            method: 'GET',
            queryFilter: filters
        };
        const response: HttpResponse = await httpClient.executeRequest(config, true);
        
        allItems = [...allItems, ...response.data.value];
        nextPageLink = response.data['@odata.nextLink'] || null;
    }

    return allItems;
}

2. 单用户多属性并行拉取 #

对每个用户，并行拉取所有需要的属性（组、角色、PIM数据等）：

async function fetchUserFullData(userId: string, httpClient: any): Promise<any> {
    // 并行发起多个属性拉取请求，每个请求都用通用分页函数处理全量数据
    const [groups, directoryRoles, pimData, signInActivity] = await Promise.all([
        fetchAllPages(`https://graph.microsoft.com/v1.0/users/${userId}/memberOf/microsoft.graph.group`, httpClient),
        fetchAllPages(`https://graph.microsoft.com/v1.0/users/${userId}/directoryRoles`, httpClient),
        fetchAllPages(`https://graph.microsoft.com/v1.0/users/${userId}/privilegedRoleAssignments`, httpClient), // 假设PIM数据接口
        fetchAllPages(`https://graph.microsoft.com/v1.0/users/${userId}/signInActivity`, httpClient)
    ]);

    // 合并所有属性到用户数据
    return {
        userId,
        groups,
        directoryRoles,
        pimData,
        signInActivity
    };
}

3. 按页处理用户数据，兼顾顺序与并行 #

在原有的用户分页逻辑中，对当前页的所有用户并行拉取全量属性，同时保证分页是顺序处理的：

async function processAllUsers(res: any, httpClient: any) {
    let nextPageLink: string | null = null;
    do {
        // 顺序拉取下一页用户基础数据
        const pagedResult: AggregationPageResult = await this.fetchNextPage(nextPageLink);
        const usersMap: Map<string, any>[] = pagedResult.data!;
        nextPageLink = pagedResult.nextPageLink!;

        // 并行处理当前页所有用户的全量属性拉取
        const userFullDataPromises = usersMap.map(user => {
            const userId = user.get('id'); // 假设用户ID存在于Map中
            return fetchUserFullData(userId, httpClient);
        });

        const userFullDataResults = await Promise.all(userFullDataPromises);

        // 合并属性数据到用户基础数据
        const mergedUsers = usersMap.map(user => {
            const fullData = userFullDataResults.find(fd => fd.userId === user.get('id'));
            return new Map([...user.entries(), ...Object.entries(fullData)]);
        });

        // 发送处理后的页数据
        this.sendPageResult(mergedUsers, res);
    } while (nextPageLink);
}

4. 资源控制优化 #

如果担心并行请求过多导致限流，可以用分批处理控制并发量：

// 分批并行处理，每批处理N个用户
async function processUsersInBatches(users: Map<string, any>[], batchSize: number, httpClient: any): Promise<any[]> {
    const results: any[] = [];
    for (let i = 0; i < users.length; i += batchSize) {
        const batch = users.slice(i, i + batchSize);
        const batchPromises = batch.map(user => fetchUserFullData(user.get('id'), httpClient));
        const batchResults = await Promise.all(batchPromises);
        results.push(...batchResults);
    }
    return results;
}

// 在分页逻辑中替换为分批处理
// const userFullDataResults = await processUsersInBatches(usersMap, 10, httpClient);

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