已在 async.map 中调用回调

Question

编辑： "threads"+"async" 模块不是异步（如果不是线程）安全的！它被另一个异步函数调用！这就是为什么在单个异步完成之前线程运行它两次。有一些代码残留异步调用相同的代码。

这是一些给出错误的代码（标有ERROR!）：

async.map(threads, function (thread, cb_) {
            zz++,
            console.log(zz);
            (function (id____) {

                thread.msg({ cmd: "compare", data: id____ }, function (result) {
                    if (result)
                        compareCheck = false;
                    console.log(cb_.toString());

                    // ERROR!
                    cb_(null, thread); // why?
                    // return cb_ doesn't solve neighter

                });
            })(id__);

        }, function (err, result) {

            if (compareCheck) {
                console.log("b:" + Session.cmpCtr);

                threads[Session.cmpCtr % nThreads].msg({
                    cmd: "add", data: id__
                }, function (response) {
                  setImmediate(  cbCmp_);
                });
            }
            else {

                // whatsoever gives collision, retry new one,
                // setImmediate to stop recursion fee on cpu
                setImmediate(function () {

                    id__ = "";
                    for (var i = 0; i < 24; i++)
                        id__ += rndChar();
                    cmpId(id__, cbCmp_);
                });
            }
        });

这段代码所做的是，检查 N 个线程（使用带有 spawn 的线程模块）在它们自己的列表中是否有一个会话变量等于新生成的一个，然后如果没有冲突，则将新变量添加到其中一个线程列表。如果线程的命令是 "add"，那么它将变量添加到它的列表中，如果命令是 "compare"，那么它根据列表的值和新值在回调中给出真假值。

为什么 async.map 会出现这个错误？该函数由所有线程执行，为什么不应该执行多次？是回调。

也许 async.map 不能有一个线程列表（它是不可序列化的）并且只是复制它以便在每次会话变量生成时有更多的多个实例？

Threads 正被初始化为以下内容的包装器使用：

threads.push(new QThread(function (input, done) {
    this.data = (typeof this.data === "undefined")? [] : this.data;
    if (input.cmd != null && typeof input.cmd !== "undefined") {
        if (input.cmd === "add") {
            data.push(input.data);
            done(true);
        }
        else if (input.cmd === "compare")
        {
            for (var i = 0; i < data.length; i++) {
                if (data[i] == input.data) {
                    done(true);
                    return;
                }
            }

            done(false);
            return;
        }
    }
}));

编辑： return cb_(null, thread); 给出同样的错误。

我想使用 async.map 因为使用简单的自旋锁样式同步是 cpu 资源密集型并且不适合 nodejs。

这是输出的错误：

C:\...\async.js:985
        if (fn === null) throw new Error("Callback was already called.");
                         ^

Error: Callback was already called.
    at C:\...\async.js:985:32
    at C:\...\async.js:1158:13
    at C:\...\session.js:79:28
    at Worker.<anonymous> (C:\...\qthread.js:25:13)
    at Worker.emit (C:\...\index.js:129:35)
    at Worker.handleMessage (C:\Users\pc\node_modules\threads\lib\worker.node\worker.js:119:17)
    at emitTwo (events.js:106:13)
    at ChildProcess.emit (events.js:191:7)
    at process.nextTick (internal/child_process.js:744:12)
    at _combinedTickCallback (internal/process/next_tick.js:67:7)

Answer 1

调用cb_(null, thread);后，函数仍然执行剩下的代码。最好像这样终止回调

return cb_(null, thread);

这确保函数调用回调、停止执行并returns。

已在 async.map 中调用回调

Callback was already called in async.map

multithreading

node.js

async.js