Nodejs中async的queue中的异常能不能不影响到全局程序的运行？

Nodejs中async的queue中的异常能不能不影响到全局程序的运行？

全局有个控制器，负责定期往queue中push任务。但子任务偶尔会抛出异常。

现在的需求是子任务异常也没关系，那就让子任务自己停掉就好，但是希望不要影响到全局的控制器。不知道这个应该怎么搞啊……

3 回复

当然可以！在Node.js中使用async库的queue时，可以通过合理处理异常来确保子任务的异常不会影响全局程序的运行。下面是一个简单的示例代码和解释，展示如何实现这一点。

示例代码

const async = require('async');

// 创建一个异步队列
const q = async.queue((task, callback) => {
    try {
        // 执行任务
        console.log(`Processing task: ${task.name}`);
        if (task.throwError) {
            throw new Error('Task encountered an error');
        }
        console.log(`Completed task: ${task.name}`);
    } catch (error) {
        console.error(`Error processing task: ${task.name}`, error.message);
    } finally {
        // 无论成功或失败，都需要调用callback来结束当前任务
        callback();
    }
}, 2);  // 设置并发数为2

// 全局控制器，定期向队列中添加任务
setInterval(() => {
    const task = { name: `Task-${Date.now()}`, throwError: Math.random() > 0.5 };
    q.push(task, (err) => {
        if (err) {
            console.error(`Failed to process task: ${task.name}`);
        } else {
            console.log(`Successfully processed task: ${task.name}`);
        }
    });
}, 1000);

console.log('Global controller is running...');

解释

  1. 创建异步队列：

    • 使用async.queue创建一个队列，并传入一个处理函数。该处理函数接收两个参数：task（当前任务）和callback（完成回调）。

  2. 任务处理函数：

    • 在任务处理函数内部，我们使用try...catch语句来捕获任何可能发生的异常。
    • 如果任务执行过程中抛出异常，我们在catch块中打印错误信息，并确保调用callback()来通知队列任务已经结束。
    • 无论任务是否成功，都需要调用callback()来结束当前任务，以防止队列阻塞。

  3. 全局控制器：

    • 使用setInterval模拟全局控制器，定期向队列中添加任务。每个任务都有一个throwError属性，用于随机决定是否抛出异常。
    • 当任务完成时，通过回调函数通知队列任务已结束，并根据结果打印相应的日志。

这样，即使某个子任务抛出异常，也不会影响全局控制器的正常运行。


先自己捕获异常，觉得可以忽略就ignore，不行就抛出。

在Node.js中使用async.queue时，如果子任务出现异常，默认情况下这些异常确实会影响到整个程序的运行。不过，你可以通过一些策略来确保子任务的异常不会影响到全局程序的运行。

解决方案

你可以在处理队列任务时捕获异常，并且在任务失败时将任务重新放入队列或者标记为完成。这样可以确保全局控制器不受子任务异常的影响。

示例代码

const async = require('async');

// 创建一个队列对象，同时指定并发数量为1
const q = async.queue((task, callback) => {
    console.log(`Processing task: ${task.name}`);
    try {
        // 模拟一个可能会抛出异常的任务
        if (task.name === 'error') throw new Error('Task failed');
        setTimeout(() => {
            console.log(`Completed task: ${task.name}`);
            callback(); // 任务完成后调用回调函数
        }, 1000);
    } catch (err) {
        console.error(`Error processing task: ${err.message}`);
        callback(err); // 将错误传递给回调函数
    }
}, 1);

// 向队列中添加任务
q.push({ name: 'normal' });
q.push({ name: 'error' });
q.push({ name: 'another normal' });

// 处理任务完成后的回调
q.drain = () => {
    console.log('All tasks have been processed');
};

// 处理任务失败后的回调
q.error = (err, task) => {
    console.error(`Task ${task.name} failed: ${err.message}`);
    // 可以选择在此处重新尝试任务，或记录日志
};

解释

  • async.queue 接收两个参数：一个处理函数和一个可选的最大并发量。
  • 在处理函数内部，我们使用了try...catch块来捕获可能的异常。
  • 如果任务抛出异常，异常被捕获后会调用callback(err)，这会让队列知道当前任务已经失败，并触发q.error回调。
  • q.error 回调提供了错误处理机制，可以在这里记录错误信息或者决定是否重新尝试该任务。
  • q.drain 回调会在所有任务都处理完毕后执行，确保全局控制器能够得知任务处理的状态。

通过这种方式，即使队列中的某些任务失败，也不会影响到全局程序的正常运行。

回到顶部