重新加载页面后，如何确保服务器和客户端上的 socket.id 相同？

Question

我正在编写使用 Node 集群（workers）、Redis 和 socket.io 的概念证明（至少 2 个月）。

在这种情况下，Socket.io 不用于聊天 - 只是从后端到前端的通信。 Ajax 不是一种选择。

我正在为 redis 使用 pub/sub 并让这部分工作（我认为）。至少从 pubClient.get('key') 返回的值是正确的。

当我从前端发出请求并且不以任何方式导航或重新加载页面时，一切正常 - 我可以发出 10 个请求并收到 10 个响应。

相反，当我导航时，情况并非如此 - 无论有人在前端导航多少，我都需要提供结果。

似乎重新加载后断开连接。在两个控制台 - 开发工具和节点 js 中，套接字 ID 是相同的。我真的在这个问题上摸不着头脑！

有什么帮助吗？

所以，对于一些主要是socket.io的代码：

客户：

    socket = io('https://' + location.hostname + ':4444/', {
        transports: ['websocket', 'polling'],
        secure: true,
    });

    socket.on('download', function(data){// after reload, this never hits
        console.log('DOWNLOAD    '+ data.download);
    });

    var pkgs = ['y14Vfk617n6j', 'My77gWYmBLxT', 'IYd6dL9UoXkx'];
    if(pkgs.length > 0){
        for(var i = 0; i < pkgs.length; i++){
            socket.emit('get-request', pkgs[i]);
        }
    }
    

服务器：

    var cluster = require('cluster');
    var express = require('express');
    var numCPUs = require('os').cpus().length;
    const { setupMaster, setupWorker } = require("@socket.io/sticky");
    const { createAdapter, setupPrimary } = require("@socket.io/cluster-adapter");
    var app = express();
    const https             = require('https');
    const { Server }        = require("socket.io");
    const Redis             = require("ioredis");
    const sock_nodes = [
        {port: 6379, host: '192.168.0.41'},
        {port: 6380, host: '192.168.0.34'},
        {port: 6381, host: '192.168.0.35'},
        {port: 6379, host: '192.168.0.34'},
        {port: 6380, host: '192.168.0.35'},
        {port: 6381, host: '192.168.0.41'}
    ];
    
    const port = 4444;
    const httpServer = https.createServer(options, app);
    const io = new Server(httpServer, {maxHttpBufferSize: 10240000});

    const pubClient = new Redis.Cluster(sock_nodes, {
        redisOptions: {
            password: 'my secret!'
        }
    });
    
    const subClient = pubClient.duplicate(); // I am not actually using this - should I be?

if (cluster.isMaster) {
    for (var i = 0; i < numCPUs; i++) {
        // Create a worker
        cluster.fork();
    }
    cluster.on("exit", (worker) => {
        console.log(`Worker PID ${worker.process.pid} died`);
        var w = cluster.fork();
        console.log('WORKER %d died (%s). restarting...', worker.process.pid, worker.state);
        w.on('message', function(msg){
            console.log("Message Received : " , msg);
        });
    });



} else {

    app.use((req, res, next) => {
        var reqip = req.headers['x-real-ip'] || req.headers['x-forwarded-for'] || req.connection.remoteAddress;
        //~ console.log(reqip, md5(reqip));

        var sess = parseCookies(req, 'session_state');
        if(!sess){
            res.cookie('session_state', md5(reqip));
        }
        next();
    });

    app.get('/', (req, res) => {
        getSession(req, res, function(sess){
          getPub('currSockets', sess, function(err, socket){
            res.render("pages/shared/index", {'ns': sess, 'socket': socket});
            });
          });
      });
    });

    app.get('/start', function(req, res){
        getSession(req, res, function(sess){
          getPub('currSockets', sess, function(err, socket){
            res.render("pages/shared/start", {'ns': sess, 'socket': socket});
          });
        });       
    });

    io.on('connection', function (socket) {
        var currUser = parseCookies(socket.request, 'session_state');
        socket.join(currUser);
        
        getPub('currSockets', currUser, function(err, currSockets){
            if (currSockets) {
                currSockets = JSON.parse(currSockets);            
                if (currSockets[currUser]) {
                    if (currSockets[currUser].stream) {                      
                        currSockets[currUser].sock = socket.id;
                        setCurrSockets(currSockets, currUser, null, function(cSocks){
                            
                        });
                    }
                }
            }
        });
        
        socket.on('get-request', function(data){ // can be one or many requests
        // there is a similar, currently irrelevant, socket.on('new-request') that is left out here           
            if(data){
                getPub('currSockets', currUser, function(err, currSockets){
                    currSockets = JSON.parse(currSockets);
                    if(currSockets){
                        if(currUser){
                            if(currSockets[currUser]){
                                if(currSockets[currUser].stream){
                                    var str = Object.keys(currSockets[currUser].stream);
                                    for(var i = 0; i < str.length; i++){
                                        if(str[i] !== 'sock'){
                                            if(!currSockets[currUser].stream[str[i]]){
                                                delete currSockets[currUser].stream[str[i]];
                                                setCurrSockets(currSockets, currUser, null, function(cSocks){                         
                                                    checkCurrSockets(currUser, data, socket);           
                                                });
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                });
            }
        });
    });

    httpServer.listen(port, () => {
        logs(__line__, `Worker ${process.pid} listening on ${port}`);
    });
}
    function existsPub(key, cb){
        return pubClient.exists(key, cb);
    }

    function setPub(key, val, cb){
        if(val === JSON.stringify({})){
            return pubClient.get(key, cb);
        }
        return pubClient.set(key, val, cb);
    }

    function getPub(key, currUser, cb){
        existsPub(key, function(err, reply){
            if(reply === 1){
                return pubClient.get(key, cb);// always getting an old socket.id
            }
        });
    }

// Here is the piece that doesn't work after reloading the page

    function ioEmit (currSock, target, payload) {
        io.to(currSock).emit(target, payload);  // doesn't work after page reload
    }
// end piece where after reload does not work


    getPub('currSockets', currUser, function(err, currSockets){
        if( currSockets){
            currSockets = JSON.parse(currSockets);
            ioEmit(currUser, 'download', {'download': currSockets[currUser].stream[data]);
        }
    });

    function parseCookies (req, name) {
        var list = {}, rc;
        rc && rc.split(';').forEach(function( cookie ) {
            var parts = cookie.split('=');
            list[parts.shift().trim()] = decodeURI(parts.join('='));
        });
        return list[name];
    }

function getSession(req, res, callback) {
    var sess = false;
    if(req.headers) {// handle  req
        var reqip = req.headers['x-real-ip'] || req.headers['x-forwarded-for'] || req.connection.remoteAddress;
        if(req.headers.cookie){
            sess = req.headers.cookie.split('=')[1].split(';')[0];
        } else {
            res.cookie('session_state', md5(reqip));
        }
        return callback(sess);

    } else if(req.request) {// handle socket
        //~ console.log('req.request.headers.cookie', req.request.headers.cookie.split('=')[1]);
        if(req.request.headers.cookie){
            sess = req.request.headers.cookie.split('=')[1].split(';')[0];
            //~ req.emit('join', sess);
            //~ callback({[sess]: {'sock': req.id}});
            callback(req.id);
        }
    } else {
        return callback(null);
    }
}

function setCurrSockets(currSockets, currUser, data, cb){
    if(Object.keys(currSockets[currUser].stream).length > 0){
        if(data){
            if(ready(currSockets, currUser, data)){
                delete currSockets[currUser].stream[data];// it appears that setCurrSockets is getting called too soon
            }
        }
        setPub('currSockets', JSON.stringify(currSockets), function(err){

        });
        if(typeof cb === 'function'){
            setTimeout(() => {
                getPub('currSockets', currUser, function(err, cSocks){
                    cb(cSocks);// updated callback to return cSocks
                }, 2000);
            });
        }
    } else {
        currSockets[currUser].stream = {};
        setPub('currSockets', JSON.stringify(currSockets), function(err){
            if(err){
 
           } else {
                if(typeof cb === 'function'){
                    cb(currSockets);// updated callback to return cSocks
                }
            }
        });
    }
}

Answer 1

想通了。问题出在这里：

for(var i = 0; i < str.length; i++){
  if(str[i] !== 'sock'){
                                            
    >>>>  if(!currSockets[currUser].stream[str[i]]){  // never true
       // delete currSockets[currUser].stream[str[i]];
       setCurrSockets(currSockets, currUser, null, function(cSocks){                         
                                                    
         checkCurrSockets(currUser, data, socket);           
       });
     }
   }
 }

所以我评论了 for 循环并保留了 setCurrSockets 部分并且它可以工作。 只是想我会分享一下，以防其他人尝试一起使用redis、节点集群和socket.io。正如@jfreind00 所说，您应该使用带有随机生成字符串的身份验证系统来存储cookie。