Async Queue Library回调传参异常：err与results始终未定义

我用async库创建了任务队列，推送任务后，期望在回调里拿到错误信息或者包含client_reference_id、stripe_customer_id的对象，但回调的err和results始终是undefined。我知道回调第一个参数是错误、第二个是返回数据，但写法没生效，求排查问题并给出正确传参方式。

队列代码：

const async = require('async');
const db = require('../models');
const logger = require('../logger');
var q = async.queue(async function(task,callback){
    console.log(`added ${task.name} to the master queue`);
    switch (task.name) {
        case 'checkout.session.completed':
            let data = task.data;
            logger.info('checkout.session.completed');
            const stripe_customer_id = data.customer;
            const client_reference_id = data.client_reference_id;
            logger.info(`Stripe Customer ID = ${stripe_customer_id}`);
            logger.info(`Client Reference ID = ${client_reference_id}`);
            try {
                await db.users.update({
                    stripe_customerID:stripe_customer_id
                },{
                    where:{
                        id:client_reference_id
                    }
                })
            } catch (error) {
                logger.error("Failure occured: " + error.message);
                callback(error.message);return;
            }
            callback(null,{
                client_reference_id:client_reference_id,
                stripe_customer_id:stripe_customer_id
            });return;
    
        default:
            break;
    }
    callback("Error: Queue task name not defined.");
},20);
module.exports = q;

任务推送代码：

req.master_q.push({name:event.type,data:data}, function(err,results){
    logger.info(`${event.type} finished...`);
    logger.info("printing results below(success unknown):");
    logger.info(JSON.stringify(results));
    if(err){
        logger.error();("Error below:");
        logger.error(err);
        res.send({
            status: "Failure",
            error_message: err
        });
    }else{
        logger.info("printing results below(successful queue):");
        logger.info(JSON.stringify(results));
        res.send({
            status: "Success",
            client_reference_id:results.client_reference_id,
            stripe_customer_id:results.stripe_customer_id
        });
    }
})

问题根源 #

你给async.queue传入的是async函数，async库会自动识别并将其当作Promise处理，此时你手动调用的callback会被完全忽略，导致推送任务的回调无法收到任何参数，所以err和results都是undefined。

正确传参方式 #

保留async函数的前提下，通过返回值传递成功结果，通过throw抛出错误来传递异常，async库会自动将这些映射到push回调的err和results参数中。

修改后的队列代码 #

const async = require('async');
const db = require('../models');
const logger = require('../logger');
var q = async.queue(async function(task){ // 移除callback参数，无需手动调用
    console.log(`added ${task.name} to the master queue`);
    switch (task.name) {
        case 'checkout.session.completed':
            let data = task.data;
            logger.info('checkout.session.completed');
            const stripe_customer_id = data.customer;
            const client_reference_id = data.client_reference_id;
            logger.info(`Stripe Customer ID = ${stripe_customer_id}`);
            logger.info(`Client Reference ID = ${client_reference_id}`);
            try {
                await db.users.update({
                    stripe_customerID:stripe_customer_id
                },{
                    where:{
                        id:client_reference_id
                    }
                });
                // 成功时返回结果，async库会自动传递给push回调
                return {
                    client_reference_id: client_reference_id,
                    stripe_customer_id: stripe_customer_id
                };
            } catch (error) {
                logger.error("Failure occured: " + error.message);
                // 抛出错误，async库会自动传递给push回调的err参数
                throw new Error(error.message);
            }
    
        default:
            // 抛出任务未定义的错误
            throw new Error("Error: Queue task name not defined.");
    }
},20);
module.exports = q;

修正任务推送代码的语法错误 #

原代码里logger.error();("Error below:");多了一个分号，导致日志无法正确输出，修正后如下：

req.master_q.push({name:event.type,data:data}, function(err,results){
    logger.info(`${event.type} finished...`);
    logger.info("printing results below(success unknown):");
    logger.info(JSON.stringify(results));
    if(err){
        logger.error("Error below:"); // 移除多余分号
        logger.error(err);
        res.send({
            status: "Failure",
            error_message: err.message // 建议取error的message属性，更规范
        });
    }else{
        logger.info("printing results below(successful queue):");
        logger.info(JSON.stringify(results));
        res.send({
            status: "Success",
            client_reference_id:results.client_reference_id,
            stripe_customer_id:results.stripe_customer_id
        });
    }
})

补充说明 #

如果一定要用传统callback方式（不推荐，与async/await混合易造成代码混乱），需要去掉任务函数的async关键字，将异步操作改为回调形式，但这种写法会降低代码可读性，因此更推荐用Promise/async-await结合async库的方式。

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