节点服务器关闭然后重新启动时出现奇怪的socket.io行为

Question

我为我的网站实现了一个简单的聊天,用户可以通过ExpressJS和Socket.io互相交流.我添加了一个简单的保护,免受ddos攻击,这可能是由一个人发送垃圾邮件引起的,如下所示:

if (RedisClient.get(user).lastMessageDate > currentTime - 1 second) {

   return error("Only one message per second is allowed")

} else {

   io.emit('message', ...)     
   RedisClient.set(user).lastMessageDate = new Date()

}

我用这段代码测试了这个:

setInterval(function() {
    $('input').val('message ' + Math.random());
    $('form').submit();
}, 1);

当节点服务器始终启动时,它可以正常工作.

但是,如果我关闭节点服务器,然后运行上面的代码,并在几秒钟内再次启动节点服务器,事情变得非常奇怪.然后突然,数百条消息被插入窗口,浏览器崩溃.我假设这是因为当Node服务器关闭时,socket.io正在保存所有客户端发出,并且一旦检测到Node服务器再次联机,它就会异步推送所有这些消息.

我怎样才能防止这种情况发生？这到底发生了什么？

编辑:如果我使用节点内存而不是Redis,这不会发生.我猜测因为服务器充满了READ并且许多READ在RedisClient.set(user).lastMessageDate = new Date()完成之前发生.我想我需要的是原子READ/SET？我正在使用此模块:https://github.com/NodeRedis/node_redis用于从Node连接到Redis.

Answer 1

你是正确的,这是由于客户端上的消息排队和服务器上泛滥而发生的.

当服务器收到消息时,它会立即收到所有消息,并且所有这些消息都不是同步的.因此,每个socket.on("message:...事件都是分开执行的,即一个事件socket.on("message...与另一个事件无关并分别执行.

即使您Redis-Server的延迟时间为几毫秒,这些消息也会立即被接收,并且所有消息都会一直处于这种else状态.

您有以下几个选项.

1. 使用像此库一样的速率限制器库.这很容易配置,并有多个配置选项.

2. 如果您想自己做所有事情,请在服务器上使用队列.这将占用您服务器上的内存,但您将实现您想要的.它不是将每条消息写入服务器,而是放入队列中.为每个新客户端创建一个新队列,并在处理队列中的最后一个项目时删除此队列.

3. (更新)使用multi + watch创建锁定,以便除当前命令之外的所有其他命令都将失败.

伪代码将是这样的.

let queue = {};

let queueHandler = user => {
  while(queue.user.length > 0){
    // your redis push logic here
  }
  delete queue.user
}


let pushToQueue = (messageObject) => {
  let user = messageObject.user;

  if(queue.messageObject.user){
    queue.user = [messageObject];
  } else {
    queue.user.push(messageObject);
  }

  queueHandler(user);
}

socket.on("message", pushToQueue(message));

UPDATE

Redis支持使用与multi一起使用的WATCH进行锁定.使用此方法,您可以锁定密钥,并且在该时间内尝试访问该密钥的任何其他命令都会失败.

来自redis客户端README

使用multi你可以确保你的修改作为一个交易运行,但你不能确定你是否先到达那里.如果另一个客户在您使用它的数据时修改了密钥怎么办？

为了解决这个问题,Redis支持WATCH命令,该命令用于与MULTI一起使用:var redis = require("redis"),client = redis.createClient({...});

client.watch("foo", function( err ){
if(err) throw err;

client.get("foo", function(err, result) {
    if(err) throw err;

    // Process result
    // Heavy and time consuming operation here

    client.multi()
        .set("foo", "some heavy computation")
        .exec(function(err, results) {

            /**
             * If err is null, it means Redis successfully attempted 
             * the operation.
             */ 
            if(err) throw err;

            /**
             * If results === null, it means that a concurrent client
             * changed the key while we were processing it and thus 
             * the execution of the MULTI command was not performed.
             * 
             * NOTICE: Failing an execution of MULTI is not considered
             * an error. So you will have err === null and results === null
             */

        });
}); });

Answer 2

也许您可以扩展客户端代码，以防止在套接字断开连接时发送数据？这样，您就可以防止库在套接字断开连接（即服务器离线）时对消息进行排队。

这可以通过检查是否socket.connected为真来实现：

// Only allow data to be sent to server when socket is connected
function sendToServer(socket, message, data) {

    if(socket.connected) {
        socket.send(message, data)
    }
}

有关这方面的更多信息可以在文档中找到：https://socket.io/docs/client-api/#socket-connected

这种方法将防止在套接字断开连接的所有情况下内置的排队行为，这可能是不可取的，但是如果应该防止您在问题中注意到的问题。

更新

或者，您可以在服务器上使用自定义中间件，通过 socket.io 的服务器 API 实现限制行为：

/*
Server side code
*/
io.on("connection", function (socket) {

    // Add custom throttle middleware to the socket when connected
    socket.use(function (packet, next) {

        var currentTime = Date.now();

        // If socket has previous timestamp, check that enough time has
        // lapsed since last message processed
        if(socket.lastMessageTimestamp) {
            var deltaTime = currentTime - socket.lastMessageTimestamp;

            // If not enough time has lapsed, throw an error back to the
            // client
            if (deltaTime < 1000) {
                next(new Error("Only one message per second is allowed"))
                return
            }
        }

        // Update the timestamp on the socket, and allow this message to
        // be processed
        socket.lastMessageTimestamp = currentTime
        next()
    });
});