socket.io 客户端在长 Node.js 函数中自动断开连接

Posted 2023-03-26

【中文标题】socket.io 客户端在长 Node.js 函数中自动断开连接

【英文标题】：socket.io client automatically disconnecting in long Node.js function

【发布时间】：2021-11-18 04:50:35

【问题描述】：

我正在使用 socket.io 将我的应用程序的 swift 客户端与服务器通信。本质上，客户端在打开应用程序时加入一个套接字连接，并且一个作业会立即添加到 Redis 队列中（这是一个需要几秒钟到大约 15 秒的作业）。服务器对作业 ID 的客户端有响应。在处理此作业时，有时客户端会断开连接。这背后似乎没有押韵或原因，因为断开连接的时间完全不一致，而且断开连接也不是发生在函数中的特定点。我想也许我是手动断开与客户端的连接，所以我在客户端每次断开连接之前设置了套接字发射（当这些发射被发射到服务器时，服务器会打印一些告诉我断开连接来自哪里的东西）。这表明断开连接是自动的，因为在结束套接字连接之前客户端从未接收到发射。这是在 Heroku 上运行的。这是我的代码：

//queue initialization
const queue = new Queue('queue', process.env.REDIS_URL)

//client pings this endpoint to get the job id in the queue
app.post('/process', async function(request, response) 
  let job = await queue(request: request.body);
  console.log("Logging job as " + job.id)
  response.json( id: job.id );
);

queue.process(10, async (job) =>  //10 is the max workers per job
    console.log("Started processing")
    const client = await pool.connect()
    let item = job.data.request
    let title = item.title
    let subtitle = item.subtitle
    let id = item.id
    io.to(id).emit("Processing1", ""); //added emissions like these because I thought maybe the socket was timing out, but this didn't help
    console.log("Processing1");

    try 
      await client.query('BEGIN')
        let geoData = await //promise of geocoding endpoint api function
        let lengthOfGeoData = geoData.context.length
        io.to(id).emit("Processing2", "");
        console.log("Processing2");
        var municipality = ""
        var area = ""
        var locality = ""
        var place = ""
        var district = ""
        var region = ""
        var country = ""
        //for loop to go through geoData and set the above values
      if (municipality != "") 
        console.log("Signing in from " + municipality + ", " + area);
       else 
        console.log("Signing in from " + area)
      
      await scrape(municipality, area, id);
      await client.query('COMMIT')
     catch(err) 
      await client.query('ROLLBACK')
      console.log(err)
    
    try 
      await client.query('BEGIN')
      const array = await //a function that queries a Postgres db for some rows, makes json objects out of them, and pushes to the 'array' variable
      var array2 = []
      for (a of array) 
        let difference = getDifference(title, subtitle, a.title, a.subtitle) //math function
        if (difference <= 10) 
          array.push(a)
        
      
      io.to(id).emit("Processing9", "");
      console.log("Processing9");
      await client.query('COMMIT')
     catch(err) 
      await client.query('ROLLBACK')
      console.log("ERROR: Failed arrayHelperFunction")
      console.log(err)
     finally 
      client.release()
      console.log("About to emit this ish to " + id) //should emit to socket here ideally to notify that the processing is done and results can be polled
      io.to(id).emit("finishedLoading", "")
      return array2;
    
);

//when the client polls the queue after it's received the 'done' notifier from the server
app.post('/poll', async function(request, response) 
  console.log("Polling")
  let id = request.body.id
  const results = await queue(id);
  for (r of results.returnvalue) 
    console.log("Sending " + r.title);
  
  response.send(results.returnvalue)
);

//scrape
async function scrape(municipality, area, id) 
  const client = await pool.connect();
  try 
    await client.query('BEGIN')
    var location = ""
    if (municipality != "") 
      location = municipality + ", " + area
     else 
      location = area
    
    let inDatabase = await client.query('SQL statement AS it_does_exist', [params]);
    io.to(id).emit("Processing3", "");
    console.log("Processing3");
    if (inDatabase.rows[0].it_does_exist == false)  
      let query = "book clubs near " + location
      var terminationTime = new Date()
      terminationTime.setHours(terminationTime.getHours() + 4);
      let date = ("0" + terminationTime.getDate()).slice(-2);
      let month = ("0" + (terminationTime.getMonth() + 1)).slice(-2);
      let year = terminationTime.getFullYear();
      let hours = terminationTime.getHours();
      let minutes = terminationTime.getMinutes();
      let seconds = terminationTime.getSeconds();
      let timestamp = year + "-" + month + "-" + date + " " + hours + ":" + minutes + ":" + seconds

      try 
        await client.query(`SQL statement`, [params]);
       catch(err) 
        console.log("FAILURE: scrape() at 1.")
        console.log(err)
      

      var queryLocation = "New York,New York,United States" //default search origination is here
      var queryGLCode = "US"
      io.to(id).emit("Processing4", "");
      console.log("Processing4");
      try 
        await fetch('https://serpapi.com/locations.json?q='+municipality+'&limit=10',  method : "GET" )
          .then(res => res.json())
          .then((json) => 
            for (let index = 0; index < 10; index++) 
              let locationAPIName = json[index].canonical_name
              let locationAPICode = json[index].country_code
              let resultLatitude = json[index].gps[1];
              let resultLongitude = json[index].gps[0];
            
          );
       catch(err) 
        console.log("FAILURE: scrape() at 2.")
        console.log(err)
      
      io.to(id).emit("Processing5", "");
      console.log("Processing5");
      try 
        await Promise.all([
          searchEvents(engine: "google_events", q: query, location: queryLocation, hl: "en", gl: queryGLCode).then(data => async function()
            try 
              await client.query('BEGIN');
              let results = data.events_results
              if (results != null) 
                console.log("first HAD results")
                for (result of results) 
                  var fixedAddress = result.address[0]
                  let address = fixedAddress + ", " + result.address[1]
                      
                  let title = result.title + address

                  var description = result.description

                  let geoData = await geocode(address); //mapbox geocode the address
                  let latitude = Number(geoData.center[0]);
                  let longitude = Number(geoData.center[1]);
                  
                    await client.query(`SQL statement`, [params]);
                  
                
                io.to(id).emit("Processing6", "");
                console.log("Processing6");
               else 
                console.log("first DID NOT have results")
              
              console.log("FIRST BLOCK")
              await client.query('COMMIT');
             catch(err) 
              console.log("Results[0] not found.")
              console.log(err)
              await client.query('ROLLBACK');
            
          ()),

          searchEvents(engine: "google_events", q: query, location: queryLocation, hl: "en", gl: queryGLCode, start: "10").then(data => async function()
            // same as the one above, just with an offset
          ()),

          searchEvents(engine: "google_events", q: query, location: queryLocation, hl: "en", gl: queryGLCode, start: "20").then(data => async function()
            // same as the one above, but with a different offset
          ())
        ])
       catch(err) 
        console.log("FAILURE: scrape() at 3.")
        console.log(err)
      

     else 
      console.log("Location already in the database.")
    
    await client.query('COMMIT')
   catch(err) 
    await client.query('ROLLBACK')
    console.log(err)
   finally 
    client.release()
    return "Resolved";
  


//Client establish socket connection
func establishConnection(_ completion: (() -> Void)? = nil) 
    let socketUrlString: String = appState.server
    self.manager = SocketManager(socketURL: URL(string: socketUrlString)!, config: [.log(false), .reconnects(true), .extraHeaders(["header": "customheader"])])
    self.socket = manager?.defaultSocket
    self.socket?.connect()
    self.socket?.once(clientEvent: .connect, callback:  (data, emitter) in
        if completion != nil
            completion!()
        
    )
  //other socket functions


//Client initial post request
func process() 
    let server = "serverstring" + "process"
    let title = "title"
    let subtitle = "subtitle"
    let package = BookPackage(title: title, subtitle: subtitle, id: mySocketID) //this is after the initial connection
    print("package is \(package)")
            
    guard let url  = URL(string: server) else  return 

    var urlRequest = URLRequest(url: url)
    
    urlRequest.addValue("application/json", forHTTPHeaderField: "Content-Type")
    urlRequest.addValue("application/json", forHTTPHeaderField: "Accept")
    
    urlRequest.httpMethod = "POST"
    
    guard let data = try? JSONEncoder().encode(package) else  return 
            
    urlRequest.httpBody = data

    let task = URLSession.shared.dataTask(with: urlRequest) 
        (data, response, error) in
        if let error = error 
            print(error)
            return
        
        guard let data = data else  return 
        guard let dataString = String(data: data, encoding: String.Encoding.utf8) else  return 
        let jsonData = Data(dataString.utf8)
        var decodedJob: Job? = nil
        do 
            decodedJob = try JSONDecoder().decode(Job.self, from: jsonData) //Job is just a struct in the same form as the json object sent back from the server
         catch 
            print(error.localizedDescription)
        
        DispatchQueue.main.async 
            self.appState.pendingJob = decodedJob
        
    
    // start the task
    task.resume()

此错误唯一一致的部分是用户断开连接之前的日志（旁注：“断开连接的原因”和“断开连接的用户”触发 socket.on('disconnect') 事件：

https://i.stack.imgur.com/7fjuU.png

https://i.stack.imgur.com/z5bmL.png

https://i.stack.imgur.com/aHNt3.png

https://i.stack.imgur.com/64WYI.png

【问题讨论】：

长函数是否阻塞了事件队列？

你能说明客户端是如何发出请求来启动这一切的吗？是来自 javascript 还是浏览器表单帖子？

您是在问题标题中使用“长函数”一词的人。请不要问我它是什么。我在问你。

那么，客户端代码在 iPhone 上运行（因为它看起来像 Swift 代码）？也许 iPhone 声称存在电源管理问题，导致它断开非活动的 socket.io 连接？这看起来可能是客户端问题。我建议搜索与 socket.io 连接断开的 ios 问题，看看你发现的无数点击中是否有任何一个看起来像你的情况。

设置正确的环境变量将自动启用服务器上的调试。对于客户端，您似乎必须设置一个 localStorage 值。这个想法是调试代码已经在 socket.io 中，您只需使用正确的设置启用它。您可能必须使用正确的设置重新启动服务器才能生效。

【参考方案1】：

你应该用 await 阻塞事件循环。客户端每隔一段时间发送一次心跳（使用pingTimeout 定义）。

由于服务器没有收到ping，所以断开连接。

你应该隔离这个过程。找到一种将它与工作/后台进程或异步一起使用的方法，另外在服务器端增加 pingTimeout 可能会对您有所帮助。

【讨论】：

我相信你有迄今为止最正确的答案（这就是我给你赏金的原因），但对我来说这不是解决方案。

@nickcoding2 我明白了，谢谢。但进展如何？你可以在代码执行开始时启动计时器吗？您可以使用 console.time(label);控制台.timeEnd();此外，您可以在每次发送心跳时进行控制台输出。

【参考方案2】：

解决你的问题的方法是在启动服务器时修改pingTimeout。

来自Socket.io：

服务器发送ping，如果客户端在pingTimeout ms内没有回复pong，则服务器认为连接已关闭。

同样，如果客户端没有收到来自服务器的 ping 在 pingInterval + pingTimeout ms 内，客户端还认为 连接已关闭。

const io = new Server(httpServer, 
  pingTimeout: 30000
);

【讨论】：

嗨，很遗憾，我已经尝试过这个解决方案，但它不起作用（尝试将 pingTimeout 设置为 50000 毫秒）。运行调试，我可以看到断开连接的原因是“客户端关闭原因传输关闭”，如果这是问题原因将是“ping 超时”。

【参考方案3】：

您可以将传输从默认的更改为：

 const io = new Server(httpServer, 
  transports: ['polling', 'websocket'],
);

这可能会解决问题，否则您也可以尝试更改upgradeTimeout 和pingTimeout