如何在GCD中实现多任务共享执行与请求批量处理？

嘿，我来帮你搞定这个问题！你现在遇到的核心问题是没有正确处理并发调用的同步逻辑——当多个线程（异步调用）同时触发你的同步函数时，可能多个实例同时启动执行，导致队列逻辑失效，本该累积的100次请求没等收集完就被提前处理了，自然就看不到预期的执行次数。

正确的实现思路 #

要实现“运行时排队，执行完清空队列”的效果，我们需要三个核心元素：

• 一个任务队列，用来存放所有等待同步结果的调用请求
• 一个状态标志，标记当前是否正在执行同步操作
• 互斥逻辑，确保同一时间只有一个同步操作在运行，所有并发调用都等待这一次同步完成后拿到结果

JavaScript 实现示例 #

class SyncManager {
  constructor() {
    this.queue = []; // 存放等待的请求resolve函数
    this.isProcessing = false; // 标记是否正在执行同步
  }

  // 你的核心同步逻辑：同步网络与数据库，返回结果
  async remoteSyncAndPush() {
    // 模拟耗时操作（比如网络请求、数据库同步）
    await new Promise(resolve => setTimeout(resolve, 100));
    console.log('完成一次同步与推送');
    return '同步完成结果';
  }

  // 对外暴露的调用方法
  async syncAndGetResult() {
    return new Promise((resolve) => {
      // 将当前请求的resolve加入队列
      this.queue.push(resolve);

      // 如果当前没有在处理，启动队列处理流程
      if (!this.isProcessing) {
        this.processQueue();
      }
    });
  }

  // 内部队列处理逻辑
  async processQueue() {
    this.isProcessing = true;
    try {
      // 只执行一次同步逻辑
      const syncResult = await this.remoteSyncAndPush();

      // 清空队列，把结果返回给所有等待的请求
      while (this.queue.length > 0) {
        const resolve = this.queue.shift();
        resolve(syncResult);
      }
    } catch (error) {
      // 处理同步过程中的错误，抛给所有等待的请求
      while (this.queue.length > 0) {
        const resolve = this.queue.shift();
        resolve(Promise.reject(error));
      }
    } finally {
      // 重置状态，允许下一批请求进入
      this.isProcessing = false;
    }
  }
}

// 测试：模拟100次并发调用
async function test() {
  const manager = new SyncManager();
  const promises = Array.from({ length: 100 }, () => manager.syncAndGetResult());
  await Promise.all(promises);
  console.log('所有100次调用均已完成');
}

test();

Python 实现示例 #

如果你用的是Python，同样可以用类似的逻辑实现：

import asyncio
from collections import deque

class SyncManager:
    def __init__(self):
        self.queue = deque() # 存放等待的Future对象
        self.is_processing = False

    # 你的核心同步逻辑
    async def remote_sync_and_push(self):
        # 模拟耗时操作
        await asyncio.sleep(0.1)
        print("完成一次同步与推送")
        return "同步完成结果"

    # 对外暴露的调用方法
    async def sync_and_get_result(self):
        # 创建Future对象存储结果
        future = asyncio.get_event_loop().create_future()
        self.queue.append(future)

        if not self.is_processing:
            # 启动队列处理任务
            asyncio.create_task(self.process_queue())
        
        return await future

    # 内部队列处理逻辑
    async def process_queue(self):
        self.is_processing = True
        try:
            # 执行一次同步逻辑
            sync_result = await self.remote_sync_and_push()

            # 处理所有等待的请求
            while self.queue:
                future = self.queue.popleft()
                if not future.done():
                    future.set_result(sync_result)
        except Exception as e:
            # 错误处理，把异常传递给所有等待的请求
            while self.queue:
                future = self.queue.popleft()
                if not future.done():
                    future.set_exception(e)
        finally:
            self.is_processing = False

# 测试：100次并发调用
async def test():
    manager = SyncManager()
    tasks = [manager.sync_and_get_result() for _ in range(100)]
    await asyncio.gather(*tasks)
    print("所有100次调用均已完成")

asyncio.run(test())

逻辑说明 #

• 不管有多少并发调用syncAndGetResult（或sync_and_get_result），都会先被加入队列
• 只有当没有正在执行的同步操作时，才会触发processQueue（或process_queue）
• processQueue中只会执行一次remoteSyncAndPush，完成后把结果返回给队列里的所有请求，确保所有并发调用都共享这一次同步的结果
• 错误处理也覆盖到了，避免队列中的请求被永久挂起

这样修改后，你测试的100次调用会触发恰好一次同步逻辑，所有调用都能拿到结果，完全符合你的预期。

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