乐闻世界logo
搜索文章和话题

如何实现 Promise 的并发控制？

2月22日 14:07

Promise 的并发控制是一个重要的性能优化技术，它允许我们限制同时执行的异步操作数量，避免资源耗尽或性能下降。

为什么需要并发控制

问题场景

javascript
// 不推荐：同时发起大量请求
async function fetchAllUrls(urls) {
  const promises = urls.map(url => fetch(url));
  const results = await Promise.all(promises);
  return results;
}

// 问题：
// 1. 可能导致浏览器或服务器资源耗尽
// 2. 网络带宽可能被占满
// 3. 可能触发服务器的限流机制
// 4. 内存占用过高

基本并发控制实现

1. 使用 p-limit 库

javascript
import pLimit from 'p-limit';

async function fetchWithLimit(urls, concurrency = 5) {
  const limit = pLimit(concurrency);
  
  const promises = urls.map(url => 
    limit(() => fetch(url))
  );
  
  const results = await Promise.all(promises);
  return results;
}

2. 手动实现并发控制

javascript
async function asyncPool(poolLimit, array, iteratorFn) {
  const ret = [];
  const executing = [];
  
  for (const item of array) {
    const p = Promise.resolve().then(() => iteratorFn(item));
    ret.push(p);
    
    if (poolLimit <= array.length) {
      const e = p.then(() => executing.splice(executing.indexOf(e), 1));
      executing.push(e);
      
      if (executing.length >= poolLimit) {
        await Promise.race(executing);
      }
    }
  }
  
  return Promise.all(ret);
}

// 使用示例
async function fetchWithConcurrency(urls, concurrency = 5) {
  return asyncPool(concurrency, urls, url => fetch(url));
}

3. 使用队列实现

javascript
class ConcurrencyControl {
  constructor(concurrency) {
    this.concurrency = concurrency;
    this.queue = [];
    this.running = 0;
  }

  async run(task) {
    if (this.running >= this.concurrency) {
      await new Promise(resolve => this.queue.push(resolve));
    }
    
    this.running++;
    
    try {
      return await task();
    } finally {
      this.running--;
      const next = this.queue.shift();
      if (next) next();
    }
  }
}

// 使用示例
async function fetchWithControl(urls, concurrency = 5) {
  const control = new ConcurrencyControl(concurrency);
  const promises = urls.map(url => 
    control.run(() => fetch(url))
  );
  return Promise.all(promises);
}

高级并发控制实现

1. 带重试机制的并发控制

javascript
class ConcurrencyControlWithRetry {
  constructor(concurrency, maxRetries = 3) {
    this.concurrency = concurrency;
    this.maxRetries = maxRetries;
    this.queue = [];
    this.running = 0;
  }

  async run(task) {
    if (this.running >= this.concurrency) {
      await new Promise(resolve => this.queue.push(resolve));
    }
    
    this.running++;
    
    let lastError;
    for (let i = 0; i < this.maxRetries; i++) {
      try {
        const result = await task();
        return result;
      } catch (error) {
        lastError = error;
        if (i < this.maxRetries - 1) {
          await new Promise(resolve => 
            setTimeout(resolve, 1000 * (i + 1))
          );
        }
      }
    }
    
    throw lastError;
  }
}

// 使用示例
async function fetchWithRetry(urls, concurrency = 5) {
  const control = new ConcurrencyControlWithRetry(concurrency, 3);
  const promises = urls.map(url => 
    control.run(() => fetch(url))
  );
  return Promise.all(promises);
}

2. 带超时的并发控制

javascript
class ConcurrencyControlWithTimeout {
  constructor(concurrency, timeout = 5000) {
    this.concurrency = concurrency;
    this.timeout = timeout;
    this.queue = [];
    this.running = 0;
  }

  async run(task) {
    if (this.running >= this.concurrency) {
      await new Promise(resolve => this.queue.push(resolve));
    }
    
    this.running++;
    
    try {
      const result = await Promise.race([
        task(),
        new Promise((_, reject) => 
          setTimeout(() => reject(new Error('Timeout')), this.timeout)
        )
      ]);
      return result;
    } finally {
      this.running--;
      const next = this.queue.shift();
      if (next) next();
    }
  }
}

// 使用示例
async function fetchWithTimeout(urls, concurrency = 5) {
  const control = new ConcurrencyControlWithTimeout(concurrency, 5000);
  const promises = urls.map(url => 
    control.run(() => fetch(url))
  );
  return Promise.all(promises);
}

3. 带优先级的并发控制

javascript
class PriorityConcurrencyControl {
  constructor(concurrency) {
    this.concurrency = concurrency;
    this.queue = [];
    this.running = 0;
  }

  async run(task, priority = 0) {
    const taskWrapper = {
      task,
      priority,
      resolve: null
    };
    
    this.queue.push(taskWrapper);
    this.queue.sort((a, b) => b.priority - a.priority);
    
    if (this.running >= this.concurrency) {
      await new Promise(resolve => {
        taskWrapper.resolve = resolve;
      });
    }
    
    this.running++;
    
    try {
      return await task();
    } finally {
      this.running--;
      const next = this.queue.find(t => t.resolve);
      if (next) {
        this.queue.splice(this.queue.indexOf(next), 1);
        next.resolve();
      }
    }
  }
}

// 使用示例
async function fetchWithPriority(urls, concurrency = 5) {
  const control = new PriorityConcurrencyControl(concurrency);
  const promises = urls.map((url, index) => 
    control.run(() => fetch(url), index % 3)
  );
  return Promise.all(promises);
}

实际应用场景

1. 批量下载文件

javascript
async function downloadFiles(urls, concurrency = 3) {
  const control = new ConcurrencyControl(concurrency);
  
  const results = await Promise.all(
    urls.map(url => 
      control.run(async () => {
        const response = await fetch(url);
        const blob = await response.blob();
        return { url, blob };
      })
    )
  );
  
  return results;
}

2. 批量处理数据库查询

javascript
async function processQueries(queries, concurrency = 5) {
  const control = new ConcurrencyControl(concurrency);
  
  const results = await Promise.all(
    queries.map(query => 
      control.run(() => database.execute(query))
    )
  );
  
  return results;
}

3. 批量发送邮件

javascript
async function sendEmails(recipients, concurrency = 10) {
  const control = new ConcurrencyControl(concurrency);
  
  const results = await Promise.allSettled(
    recipients.map(recipient => 
      control.run(() => emailService.send(recipient))
    )
  );
  
  const successful = results.filter(r => r.status === 'fulfilled').length;
  const failed = results.filter(r => r.status === 'rejected').length;
  
  console.log(`发送完成: 成功 ${successful}, 失败 ${failed}`);
  
  return results;
}

性能优化技巧

1. 动态调整并发数

javascript
class AdaptiveConcurrencyControl {
  constructor(initialConcurrency = 5, maxConcurrency = 20) {
    this.concurrency = initialConcurrency;
    this.maxConcurrency = maxConcurrency;
    this.minConcurrency = 1;
    this.queue = [];
    this.running = 0;
    this.successCount = 0;
    this.errorCount = 0;
  }

  async run(task) {
    if (this.running >= this.concurrency) {
      await new Promise(resolve => this.queue.push(resolve));
    }
    
    this.running++;
    
    try {
      const result = await task();
      this.successCount++;
      this.adjustConcurrency();
      return result;
    } catch (error) {
      this.errorCount++;
      this.adjustConcurrency();
      throw error;
    } finally {
      this.running--;
      const next = this.queue.shift();
      if (next) next();
    }
  }

  adjustConcurrency() {
    const total = this.successCount + this.errorCount;
    const errorRate = this.errorCount / total;
    
    if (errorRate < 0.1 && this.concurrency < this.maxConcurrency) {
      this.concurrency = Math.min(this.concurrency + 1, this.maxConcurrency);
    } else if (errorRate > 0.3 && this.concurrency > this.minConcurrency) {
      this.concurrency = Math.max(this.concurrency - 1, this.minConcurrency);
    }
  }
}

2. 进度监控

javascript
class ConcurrencyControlWithProgress {
  constructor(concurrency, onProgress) {
    this.concurrency = concurrency;
    this.onProgress = onProgress;
    this.queue = [];
    this.running = 0;
    this.completed = 0;
    this.total = 0;
  }

  async run(task) {
    this.total++;
    
    if (this.running >= this.concurrency) {
      await new Promise(resolve => this.queue.push(resolve));
    }
    
    this.running++;
    
    try {
      const result = await task();
      return result;
    } finally {
      this.running++;
      this.completed++;
      this.onProgress(this.completed, this.total);
      
      this.running--;
      const next = this.queue.shift();
      if (next) next();
    }
  }
}

// 使用示例
async function fetchWithProgress(urls, concurrency = 5) {
  const control = new ConcurrencyControlWithProgress(concurrency, (completed, total) => {
    console.log(`进度: ${completed}/${total} (${(completed/total*100).toFixed(1)}%)`);
  });
  
  const promises = urls.map(url => 
    control.run(() => fetch(url))
  );
  
  return Promise.all(promises);
}

常见问题

1. 如何选择合适的并发数？

javascript
// 根据网络类型选择
function getOptimalConcurrency() {
  const connection = navigator.connection || navigator.mozConnection || navigator.webkitConnection;
  if (connection) {
    return Math.min(connection.downlink || 4, 10);
  }
  return 4; // 默认值
}

2. 如何处理失败的任务？

javascript
async function fetchWithPartialFailure(urls, concurrency = 5) {
  const control = new ConcurrencyControl(concurrency);
  
  const results = await Promise.allSettled(
    urls.map(url => 
      control.run(() => fetch(url))
    )
  );
  
  const successful = results.filter(r => r.status === 'fulfilled');
  const failed = results.filter(r => r.status === 'rejected');
  
  console.log(`成功: ${successful.length}, 失败: ${failed.length}`);
  
  return { successful, failed };
}

总结

  1. 避免资源耗尽：限制同时执行的异步操作数量
  2. 提高性能：合理的并发控制可以提高整体性能
  3. 灵活实现：可以根据需求实现不同的并发控制策略
  4. 错误处理：结合重试、超时等机制提高可靠性
  5. 进度监控：实时监控任务执行进度
  6. 动态调整：根据实际情况动态调整并发数
标签：Promise
热门标签
更多
Git(114)C语言(23)C++(15)React(34)前端(148)Cypress(27)JavaScript(56)Linux(20)Rust(10)Docker(65)Mongoose(18)TypeScript(27)MySQL(8)Vue(13)Tailwind CSS(29)CSS(19)ElasticSearch(24)网络(12)TypeORM(31)