javascript - Node.js redis pub/sub 丢失消息

Question

我试图从两天开始调试这段代码，但我无法弄清楚所以我问你。
代码非常简单:两个端点做几乎相同的事情。
第一个，“s”，监听 channel “idS”上的任何消息。如果 10 秒后它没有收到任何东西，它就会结束。如果 req.body 不为空，则将其发送给 'c'。
在第二个中，'c'，始终向'idS'发送消息并等待 channel 'idC'上的消息。
在“s”中，变量“id”设置为 1，因此“c”可以了解“s”何时在线。
我有两个不同的测试程序(每个端点一个)。
在 testS 中，我在没有 req.body 的情况下调用端点，当我收到答案时，我再次调用端点，这次是使用 req.body。
在 testC 中，我只是调用端点并等待答案。当我收到它时，我会在一秒钟后再次调用端点。

测试应该以这种方式工作:

1. s connect，除了在 channel “idS”上监听外，它没有什么可发送的。
2. c 连接，在 channel “idC”上收听并在“idS”上发送“测试”。
3. s 在 channel “idS”上收到消息“test”并退出。
4. 一个新的 s 连接，在 channel “idS”上收听并向“idC”上的 c 发送消息。
5. c 在 channel “idC”上接收并退出。
6. 新的 c 在一秒后再次连接并从 2 开始重复。

var express = require('express'),
    redis = require('redis'),
    util = require('util'),
    client = redis.createClient(6379, 'localhost');
var router = express.Router();

router.post('/s', function(req, res){
    var id = req.query.code;
    var cId = "c_" + id;
    var sId= "s_" + id;

    //BUILD RESPONS
    var respons = {....};

    var clientSub = redis.createClient(6379, "localhost");
    clientSub.on("message", function(channel, msg){
        if(timeoutId)
            clearTimeout(timeoutId);
        client.del(id);
        respons.val = idS;
        clientSub.unsubscribe(idS);
        clientSub.quit();
        res.send(respons).end();
    });
    clientSub.subscribe(idS);
    client.set(id, 1);

    if(req.body){
        client.publish(idC, JSON.stringify(req.body));
    }

    var timeoutId = setTimeout(function(){
        if(!res.headersSent){
            clientSub.unsubscribe(idS);
            clientSub.quit();
            client.del(id);
            console.log(" HUB TIMEOUTED " + id);

            res.json(respons).end();
        }
    },1000*10);
});

router.post('/c', function(req, res){    
    var id = //code to get id
    var idC = "c_" + id;
    var idS= "s_" + id;


    var clientSub = redis.createClient(6379, "localhost");
    clientSub.once("message", function(channel, msg){
        var respons = {};
        respons.data = msg;
        clientSub.unsubscribe(idC);
        clientSub.quit();
        res.json(respons).end();
    });
    //subscribe to a channel
    clientSub.subscribe(idC);


    //try to send message
    client.get(id, function(err, reply){
        if(err)
            console.log(err);
        else if(reply == 1){
            client.publish(idS, "test");
        } else{
            console.log("ERROR!!!");
        }

    });
});

显然，问题是“c”有时没有收到“idC”上的消息(或者 s 没有发送消息？？)。
该错误不在测试程序中。

感谢您的帮助!

Answer 1

用 then-redis 解决。问题出在同步问题上。