异步队列库，将变量传递给回调

Question

我正在使用异步库来创建队列。

我的队列代码看起来像这样。

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;

我要做的是传回错误消息或包含 client_reference_id 或 stripe_customer_id 的对象。

我将任务推入队列的代码是这样的

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
                    });
                }
            })

出于某种原因，错误和结果总是空的。我可以通过控制台判断，如下图。

显然，错误和结果都是未定义的，这应该是不可能的，因为回调代码显然只允许回调以一种或另一种方式返回某些东西，而且我知道它正确地获得了 client_reference_id，因为它显示在控制台中。

我的猜测是我误解了图书馆的工作原理在回调中传回变量但是文档缺少这方面的信息，我在网上找不到任何东西。

我的理解是回调第一个参数是错误，第二个是您可能想要发回的数据，或者回调必须与您在创建任务时使用的任何回调函数的格式相匹配？我不确定，因为不清楚。也许这两者都不是？

特别是这里的代码应该可以工作

callback(null,{
         client_reference_id:client_reference_id,
         stripe_customer_id:stripe_customer_id
});return;

我的理解是通过为第一个参数发送 null 没有错误，我可以发送一个包含我想在第二个参数中发送回的对象，但它似乎不起作用！

此处链接到文档：https://caolan.github.io/async/v3/docs.html#queue

Answer 1

答案是您不能对（任务，回调）的定义进行异步因为这将在处理回调的动态生成的方式中产生错误。

要解决此问题，您可以执行以下操作，在定义中创建一个自调用异步函数。

const async = require('async');
const db = require('../models');
const logger = require('../logger');

var q = async.queue((task,callback) => { //@JA - Putting async on the queue definition function will create errors with callback. DO NOT DO.
    (async () => {
        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
                        }
                    });
                    logger.info("Successfully updated user if it existed");
                } catch (error) {
                    logger.error("Failure occured: " + error.message);
                    callback(error.message,null);return;
                }
                logger.info("Calling success callback below");
                callback(null,{client_reference_id:client_reference_id,stripe_customer_id:stripe_customer_id});return;
            default:
                break;
        }
        callback("Error: Queue task name not defined.",null);;
    })();
},20);

module.exports = q;

从 push 方法中使用它的示例：

req.master_q.push({name:event.type,data:data}, (err,results) => {
                logger.info(`${event.type} finished...`);
                // logger.info("printing results below(success unknown):");
                // logger.info(JSON.stringify(err));
                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(`results=${results}`);
                    logger.info(JSON.stringify(results));
                    res.send({
                        status: "Success",
                        client_reference_id:results.client_reference_id,
                        stripe_customer_id:results.stripe_customer_id
                    });
                }
            })