异步流程控制库

异步流程控制库的必要性

Node.js的核心特性是非阻塞I/O，这使得异步编程成为常态。回调函数虽然简单，但在处理复杂异步逻辑时容易出现"回调地狱"。多层嵌套的回调不仅难以阅读，错误处理也变得复杂。异步流程控制库应运而生，它们提供了更优雅的方式来组织异步代码。

// 回调地狱示例
fs.readFile('file1.txt', (err, data1) => {
  if (err) throw err;
  fs.readFile('file2.txt', (err, data2) => {
    if (err) throw err;
    fs.writeFile('output.txt', data1 + data2, (err) => {
      if (err) throw err;
      console.log('完成!');
    });
  });
});

常见的异步流程控制模式

串行执行

多个异步任务按顺序执行，前一个任务完成后才开始下一个。这种模式适用于有依赖关系的任务。

// 使用async库实现串行执行
const async = require('async');

async.series([
  callback => fs.readFile('file1.txt', 'utf8', callback),
  callback => fs.readFile('file2.txt', 'utf8', callback),
  callback => fs.readFile('file3.txt', 'utf8', callback)
], (err, results) => {
  if (err) throw err;
  console.log(results); // 三个文件内容的数组
});

并行执行

多个异步任务同时启动，等待所有任务完成。这种模式适合无依赖关系的任务，可以提高执行效率。

// 使用async库实现并行执行
async.parallel([
  callback => fs.readFile('file1.txt', 'utf8', callback),
  callback => fs.readFile('file2.txt', 'utf8', callback),
  callback => fs.readFile('file3.txt', 'utf8', callback)
], (err, results) => {
  if (err) throw err;
  console.log(results); // 三个文件内容的数组
});

有限制的并行执行

控制同时执行的异步任务数量，避免资源耗尽。这在处理大量任务时特别有用。

// 使用async.queue实现有限制的并行
const queue = async.queue((task, callback) => {
  console.log('处理任务:', task);
  setTimeout(() => {
    callback();
  }, 1000);
}, 2); // 同时最多2个任务

// 添加任务
for (let i = 0; i < 10; i++) {
  queue.push(i);
}

queue.drain = () => {
  console.log('所有任务完成');
};

主流异步流程控制库

Async.js

Async.js是最流行的异步流程控制库之一，提供了丰富的控制流方法。

// 使用async.waterfall实现依赖传递
async.waterfall([
  callback => fs.readFile('file1.txt', 'utf8', (err, data) => callback(err, data)),
  (data1, callback) => fs.readFile('file2.txt', 'utf8', (err, data2) => callback(err, data1, data2)),
  (data1, data2, callback) => fs.writeFile('output.txt', data1 + data2, callback)
], (err) => {
  if (err) throw err;
  console.log('完成!');
});

Bluebird

Bluebird是一个功能强大的Promise库，提供了丰富的Promise扩展功能。

const Promise = require('bluebird');
const fs = Promise.promisifyAll(require('fs'));

// 使用Promise实现串行执行
fs.readFileAsync('file1.txt', 'utf8')
  .then(data1 => fs.readFileAsync('file2.txt', 'utf8').then(data2 => [data1, data2]))
  .then(([data1, data2]) => fs.writeFileAsync('output.txt', data1 + data2))
  .then(() => console.log('完成!'))
  .catch(err => console.error(err));

co

co库基于Generator函数，可以用同步的方式写异步代码。

const co = require('co');
const fs = require('fs');
const util = require('util');
const readFile = util.promisify(fs.readFile);
const writeFile = util.promisify(fs.writeFile);

co(function* () {
  const data1 = yield readFile('file1.txt', 'utf8');
  const data2 = yield readFile('file2.txt', 'utf8');
  yield writeFile('output.txt', data1 + data2);
  console.log('完成!');
}).catch(err => console.error(err));

现代JavaScript中的异步控制

async/await

ES2017引入的async/await语法让异步代码看起来像同步代码。

async function processFiles() {
  try {
    const data1 = await fs.promises.readFile('file1.txt', 'utf8');
    const data2 = await fs.promises.readFile('file2.txt', 'utf8');
    await fs.promises.writeFile('output.txt', data1 + data2);
    console.log('完成!');
  } catch (err) {
    console.error(err);
  }
}

processFiles();

Promise.all

处理多个并行Promise的简单方法。

async function processInParallel() {
  try {
    const [data1, data2] = await Promise.all([
      fs.promises.readFile('file1.txt', 'utf8'),
      fs.promises.readFile('file2.txt', 'utf8')
    ]);
    await fs.promises.writeFile('output.txt', data1 + data2);
    console.log('完成!');
  } catch (err) {
    console.error(err);
  }
}

processInParallel();

错误处理策略

回调风格

async.waterfall([
  callback => fs.readFile('file1.txt', 'utf8', callback),
  (data1, callback) => fs.readFile('file2.txt', 'utf8', (err, data2) => {
    if (err) return callback(new Error('读取file2失败'));
    callback(null, data1, data2);
  })
], (err, data1, data2) => {
  if (err) return console.error('处理失败:', err.message);
  console.log('成功:', data1, data2);
});

Promise风格

fs.promises.readFile('file1.txt', 'utf8')
  .then(data1 => fs.promises.readFile('file2.txt', 'utf8').then(data2 => [data1, data2]))
  .then(([data1, data2]) => fs.promises.writeFile('output.txt', data1 + data2))
  .catch(err => {
    if (err.code === 'ENOENT') {
      console.error('文件不存在');
    } else {
      console.error('未知错误:', err);
    }
  });

async/await风格

async function processWithRetry() {
  let retries = 3;
  while (retries--) {
    try {
      const data = await fs.promises.readFile('file.txt', 'utf8');
      return data;
    } catch (err) {
      if (retries === 0) throw err;
      await new Promise(resolve => setTimeout(resolve, 1000));
    }
  }
}

processWithRetry().catch(err => console.error('最终失败:', err));

高级控制流模式

自动重试

async function retry(fn, retries = 3, delay = 1000) {
  try {
    return await fn();
  } catch (err) {
    if (retries <= 0) throw err;
    await new Promise(resolve => setTimeout(resolve, delay));
    return retry(fn, retries - 1, delay);
  }
}

retry(() => fs.promises.readFile('file.txt', 'utf8'))
  .then(data => console.log(data))
  .catch(err => console.error('读取失败:', err));

超时控制

function withTimeout(promise, timeout) {
  return Promise.race([
    promise,
    new Promise((_, reject) => 
      setTimeout(() => reject(new Error('操作超时')), timeout)
  ]);
}

withTimeout(fs.promises.readFile('large.txt', 'utf8'), 2000)
  .then(data => console.log(data))
  .catch(err => console.error(err.message));

并发控制

async function parallelWithLimit(tasks, limit) {
  const results = [];
  const executing = new Set();
  
  for (const task of tasks) {
    const p = task().then(result => {
      executing.delete(p);
      return result;
    });
    
    executing.add(p);
    results.push(p);
    
    if (executing.size >= limit) {
      await Promise.race(executing);
    }
  }
  
  return Promise.all(results);
}

const tasks = Array(10).fill().map((_, i) => 
  () => fs.promises.writeFile(`file${i}.txt`, `内容${i}`));

parallelWithLimit(tasks, 3)
  .then(() => console.log('所有文件写入完成'))
  .catch(err => console.error(err));

性能考量

异步流程控制库的选择会影响应用性能。Async.js在大量简单任务时表现良好，而Bluebird的Promise实现比原生Promise更快。现代JavaScript引擎不断优化async/await性能，使其成为大多数场景的最佳选择。

// 性能测试示例
const benchmark = async () => {
  // 测试async/await
  console.time('async/await');
  for (let i = 0; i < 1000; i++) {
    await Promise.resolve(i);
  }
  console.timeEnd('async/await');

  // 测试Promise.then
  console.time('Promise.then');
  let chain = Promise.resolve();
  for (let i = 0; i < 1000; i++) {
    chain = chain.then(() => Promise.resolve(i));
  }
  await chain;
  console.timeEnd('Promise.then');
};

benchmark();

与其他Node.js特性集成

异步流程控制库常与Stream、EventEmitter等Node.js核心API结合使用。

// 结合Stream和Promise
function streamToPromise(stream) {
  return new Promise((resolve, reject) => {
    stream.on('end', resolve);
    stream.on('error', reject);
  });
}

async function processStream() {
  const readStream = fs.createReadStream('input.txt');
  const writeStream = fs.createWriteStream('output.txt');
  
  readStream.pipe(writeStream);
  await streamToPromise(writeStream);
  console.log('流处理完成');
}

processStream().catch(err => console.error(err));

自定义流程控制工具

根据特定需求创建自定义控制流工具可以更好地满足项目需求。

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

  push(task) {
    this.queue.push(task);
    this.next();
  }

  next() {
    while (this.running < this.concurrency && this.queue.length) {
      const task = this.queue.shift();
      task().finally(() => {
        this.running--;
        this.next();
      });
      this.running++;
    }
  }
}

// 使用自定义队列
const queue = new TaskQueue(2);
for (let i = 0; i < 5; i++) {
  queue.push(() => new Promise(resolve => 
    setTimeout(() => {
      console.log(`任务 ${i} 完成`);
      resolve();
    }, 1000)
  ));
}