Niezawodność transportu Websocket (utrata danych Socket.io podczas ponownego połączenia)

Używany

NodeJS, Socket.io

Problem

Wyobraź sobie, że jest dwóch użytkowników U1 i U2 połączonych z aplikacją przez Socket.io. Algorytm jest następujący:

1. U1 całkowicie traci połączenie z Internetem (np. Wyłącza Internet)
2. U2 wysyła wiadomość do U1 .
3. U1 nie otrzymuje jeszcze wiadomości, ponieważ Internet nie działa
4. Serwer wykrywa rozłączenie U1 na podstawie limitu czasu pulsu
5. U1 ponownie łączy się z socket.io
6. U1 nigdy nie otrzymuje wiadomości od U2 - chyba zgubiła się w kroku 4.

Możliwe wyjaśnienie

Myślę, że rozumiem, dlaczego tak się dzieje:

• Krok 4 na serwer zabija instancji gnieździe i kolejkę wiadomości do U1 oraz
• Ponadto w kroku 5 U1 i Serwer tworzą nowe połączenie (nie jest ono ponownie wykorzystywane), więc nawet jeśli wiadomość nadal znajduje się w kolejce, poprzednie połączenie i tak zostaje utracone.

Potrzebuję pomocy

Jak mogę zapobiec tego rodzaju utracie danych? Muszę używać beatów, bo ludzie nie wiszą na zawsze w aplikacji. Muszę też nadal dawać możliwość ponownego połączenia, ponieważ wdrażając nową wersję aplikacji, nie chcę żadnych przestojów.

PS To, co nazywam „wiadomością”, to nie tylko wiadomość tekstowa, którą mogę przechowywać w bazie danych, ale wartościowa wiadomość systemowa, której dostarczenie musi być zagwarantowane, w przeciwnym razie interfejs użytkownika spieprzy.

Dzięki!

Dodatek 1

Mam już system kont użytkowników. Co więcej, moja aplikacja jest już złożona. Dodawanie statusów offline / online nie pomoże, ponieważ mam już tego typu rzeczy. Problem jest inny.

Sprawdź krok 2. Na tym etapie technicznie nie możemy powiedzieć, czy U1 przejdzie w tryb offline , po prostu traci połączenie, powiedzmy na 2 sekundy, prawdopodobnie z powodu złego internetu. Więc U2 wysyła mu wiadomość, ale U1 jej nie otrzymuje, ponieważ internet nadal nie działa (krok 3). Krok 4 jest potrzebny do wykrycia użytkowników offline, powiedzmy, limit czasu wynosi 60 sekund. W końcu w ciągu kolejnych 10 sekund połączenie internetowe dla U1 jest aktywne i ponownie łączy się z socket.io. Ale wiadomość z U2 została utracona w przestrzeni, ponieważ na serwerze U1 została rozłączona przez przekroczenie limitu czasu.

To jest problem, nie chciałbym dostarczać w 100%.

Rozwiązanie

1. Zbierz emisję (nazwę emisji i dane) w użytkowniku {}, identyfikowanym przez losowy identyfikator emitID. Wyślij emisję
2. Potwierdź emisję po stronie klienta (wyślij emisję z powrotem do serwera z emitID)
3. Jeśli potwierdzone - usuń obiekt z {} identyfikowany przez emitID
4. Jeśli użytkownik ponownie się połączył - sprawdź {} dla tego użytkownika i przejdź przez niego, wykonując krok 1 dla każdego obiektu w {}
5. Po rozłączeniu lub / i podłączeniu w razie potrzeby spłukać {} dla użytkownika

// Server
const pendingEmits = {};

socket.on('reconnection', () => resendAllPendingLimits);
socket.on('confirm', (emitID) => { delete(pendingEmits[emitID]); });

// Client
socket.on('something', () => {
    socket.emit('confirm', emitID);
});

Rozwiązanie 2 (trochę)

Dodano 1 lutego 2020 r.

Chociaż nie jest to tak naprawdę rozwiązanie dla Websockets, komuś może się przydać. Przeprowadziliśmy migrację z Websockets do SSE + Ajax. SSE umożliwia łączenie się z klientem w celu utrzymania trwałego połączenia TCP i odbierania wiadomości z serwera w czasie rzeczywistym. Aby wysłać wiadomości od klienta na serwer - po prostu użyj Ajax. Istnieją wady, takie jak opóźnienia i narzuty, ale SSE gwarantuje niezawodność, ponieważ jest to połączenie TCP.

Ponieważ używamy Express, używamy tej biblioteki do SSE https://github.com/dpskvn/express-sse , ale możesz wybrać tę, która Ci odpowiada.

SSE nie jest obsługiwane w IE i większości wersji Edge, więc potrzebujesz polyfill: https://github.com/Yaffle/EventSource .

Inni zasugerowali to w innych odpowiedziach i komentarzach, ale główny problem polega na tym, że Socket.IO jest tylko mechanizmem dostarczania i nie można polegać na nim samym w celu zapewnienia niezawodnej dostawy. Jedyną osobą, która wie na pewno, że wiadomość została pomyślnie dostarczona do klienta, jest sam klient . W przypadku tego rodzaju systemu zalecałbym wykonanie następujących stwierdzeń:

1. Wiadomości nie są wysyłane bezpośrednio do klientów; zamiast tego są wysyłane na serwer i przechowywane w jakimś magazynie danych.
2. Klienci są odpowiedzialni za pytanie „co przegapiłem” po ponownym połączeniu i będą wysyłać zapytania do przechowywanych wiadomości w magazynie danych, aby zaktualizować ich stan.
3. Jeśli wiadomość zostanie wysłana do serwera, gdy klient będący odbiorcą jest podłączony, zostanie wysłana w czasie rzeczywistym do klienta.

Oczywiście, w zależności od potrzeb twojej aplikacji, możesz dostroić fragmenty tego - na przykład możesz użyć, powiedzmy, listy Redis lub posortowanego zestawu wiadomości i wyczyścić je, jeśli wiesz na pewno, że klient jest włączony spotykać się z kimś.

Oto kilka przykładów:

Szczęśliwa ścieżka :

• U1 i U2 są podłączone do systemu.
• U2 wysyła wiadomość do serwera, którą powinien otrzymać U1.
• Serwer przechowuje wiadomość w jakimś trwałym magazynie, oznaczając ją dla U1 jakimś znacznikiem czasu lub sekwencyjnym identyfikatorem.
• Serwer wysyła wiadomość do U1 przez Socket.IO.
• Klient U1 potwierdza (być może przez wywołanie zwrotne Socket.IO), że otrzymał wiadomość.
• Serwer usuwa utrwalony komunikat z magazynu danych.

Ścieżka offline :

• U1 traci łączność z Internetem.
• U2 wysyła wiadomość do serwera, którą powinien otrzymać U1.
• Serwer przechowuje wiadomość w jakimś trwałym magazynie, oznaczając ją dla U1 jakimś znacznikiem czasu lub sekwencyjnym identyfikatorem.
• Serwer wysyła wiadomość do U1 przez Socket.IO.
• Klient U1 nie potwierdza odbioru, ponieważ jest offline.
• Być może U2 wysyła U1 jeszcze kilka wiadomości; wszystkie są przechowywane w magazynie danych w ten sam sposób.
• Kiedy U1 łączy się ponownie, pyta serwer „Ostatnia wiadomość, którą widziałem, to X / Mam stan X, co przegapiłem”.
• Serwer wysyła do U1 wszystkie pominięte komunikaty z magazynu danych na podstawie żądania U1
• Klient U1 potwierdza odbiór, a serwer usuwa te wiadomości z magazynu danych.

Jeśli absolutnie chcesz mieć gwarantowaną dostawę, ważne jest, aby zaprojektować system w taki sposób, aby połączenie nie miało znaczenia, a dostawa w czasie rzeczywistym była po prostu bonusem ; prawie zawsze dotyczy to pewnego rodzaju magazynu danych. Jak wspomniał użytkownik568109 w komentarzu, istnieją systemy przesyłania wiadomości, które ograniczają przechowywanie i dostarczanie wspomnianych wiadomości, i może warto przyjrzeć się takiemu gotowemu rozwiązaniu. (Prawdopodobnie nadal będziesz musiał samodzielnie napisać integrację Socket.IO).

Jeśli nie jesteś zainteresowany przechowywaniem wiadomości w bazie danych, możesz uciec od przechowywania ich w lokalnej tablicy; serwer próbuje wysłać wiadomość do U1 i przechowuje ją na liście "wiadomości oczekujących", dopóki klient U1 nie potwierdzi, że ją otrzymał. Jeśli klient jest w trybie offline, po powrocie może powiedzieć serwerowi „Hej, zostałem rozłączony, wyślij mi wszystko, co przegapiłem”, a serwer może iterować po tych wiadomościach.

Na szczęście Socket.IO zapewnia mechanizm, który pozwala klientowi „odpowiedzieć” na wiadomość, która wygląda jak natywne wywołania zwrotne JS. Oto pseudokod:

// server
pendingMessagesForSocket = [];

function sendMessage(message) {
  pendingMessagesForSocket.push(message);
  socket.emit('message', message, function() {
    pendingMessagesForSocket.remove(message);
  }
};

socket.on('reconnection', function(lastKnownMessage) {
  // you may want to make sure you resend them in order, or one at a time, etc.
  for (message in pendingMessagesForSocket since lastKnownMessage) {
    socket.emit('message', message, function() {
      pendingMessagesForSocket.remove(message);
    }
  }
});

// client
socket.on('connection', function() {
  if (previouslyConnected) {
    socket.emit('reconnection', lastKnownMessage);
  } else {
    // first connection; any further connections means we disconnected
    previouslyConnected = true;
  }
});

socket.on('message', function(data, callback) {
  // Do something with `data`
  lastKnownMessage = data;
  callback(); // confirm we received the message
});

Jest to dość podobne do ostatniej sugestii, po prostu bez trwałego magazynu danych.

Możesz być także zainteresowany koncepcją pozyskiwania zdarzeń .

Odpowiedź Michelle jest dość trafna, ale należy wziąć pod uwagę kilka innych ważnych kwestii. Główne pytanie, które należy sobie zadać, brzmi: „Czy istnieje różnica między użytkownikiem a gniazdem w mojej aplikacji?” Inny sposób zadawania pytań brzmi: „Czy każdy zalogowany użytkownik może mieć więcej niż jedno połączenie przez gniazdo w tym samym czasie?”

W świecie sieci WWW prawdopodobnie zawsze istnieje możliwość, że jeden użytkownik ma wiele połączeń przez gniazdo, chyba że specjalnie wprowadzono coś, co temu zapobiega. Najprostszym przykładem jest sytuacja, w której użytkownik ma otwarte dwie karty tej samej strony. W takich przypadkach nie zależy Ci na wysłaniu wiadomości / zdarzenia do użytkownika tylko raz ... musisz wysłać go do każdej instancji gniazda dla tego użytkownika, aby każda karta mogła uruchomić wywołania zwrotne w celu zaktualizowania stanu interfejsu użytkownika. Może nie dotyczy to niektórych zastosowań, ale moje przeczucia mówią, że dotyczyłoby to większości. Jeśli Cię to niepokoi, czytaj dalej ....

Aby rozwiązać ten problem (zakładając, że używasz bazy danych jako pamięci trwałej) potrzebujesz 3 tabel.

1. użytkownicy - czyli 1 do 1 z prawdziwymi ludźmi
2. klienci - co reprezentuje „kartę”, która może mieć pojedyncze połączenie z serwerem gniazda. (każdy „użytkownik” może mieć wiele)
3. wiadomości - wiadomość, która musi zostać wysłana do klienta (nie wiadomość, która ma zostać wysłana do użytkownika lub do gniazda)

Tabela użytkowników jest opcjonalna, jeśli Twoja aplikacja jej nie wymaga, ale OP powiedział, że ją ma.

Inną rzeczą, którą należy odpowiednio zdefiniować, jest „co to jest połączenie przez gniazdo?”, „Kiedy tworzone jest połączenie z gniazdem?”, „Kiedy jest ponownie używane połączenie z gniazdem?”. Psudocode Michelle sprawia, że ​​wygląda na to, że połączenie przez gniazdo może być ponownie użyte. Dzięki Socket.IO NIE MOŻNA ich ponownie użyć. Widziałem, że jestem źródłem wielu nieporozumień. Istnieją scenariusze z życia, w których przykład Michelle ma sens. Ale muszę sobie wyobrazić, że te scenariusze są rzadkie. To, co naprawdę się dzieje, to utrata połączenia przez gniazdo, to połączenie, identyfikator itp. Nigdy nie zostaną ponownie wykorzystane. Tak więc wszelkie wiadomości oznaczone specjalnie dla tego gniazda nigdy nie zostaną nikomu dostarczone, ponieważ gdy klient, który pierwotnie się połączył, ponownie się połączy, otrzyma zupełnie nowe połączenie i nowy identyfikator. To znaczy to ”

Tak więc dla przykładu opartego na Internecie tutaj byłby zestaw kroków, które polecam:

• Kiedy użytkownik ładuje klienta (zazwyczaj pojedynczą stronę internetową), który ma potencjał tworzenia połączenia przez gniazdo, dodaj wiersz do bazy danych klienta, który jest powiązany z jego identyfikatorem użytkownika.
• Kiedy użytkownik faktycznie łączy się z serwerem gniazda, przekaż identyfikator klienta do serwera z żądaniem połączenia.
• Serwer powinien sprawdzić, czy użytkownik może się połączyć, a wiersz klienta w tabeli klientów jest dostępny do połączenia i odpowiednio zezwolić / odmówić.
• Zaktualizuj wiersz klienta o identyfikator gniazda wygenerowany przez Socket.IO.
• Wyślij wszystkie elementy w tabeli wiadomości połączone z identyfikatorem klienta. Nie byłoby żadnego przy początkowym połączeniu, ale jeśli pochodziło to od klienta próbującego ponownie się połączyć, może być.
• Za każdym razem, gdy trzeba wysłać wiadomość do tego gniazda, dodaj wiersz w tabeli komunikatów, który jest powiązany z wygenerowanym identyfikatorem klienta (nie identyfikatorem gniazda).
• Spróbuj wyemitować wiadomość i nasłuchuj klienta z potwierdzeniem.
• Po otrzymaniu potwierdzenia usuń ten element z tabeli komunikatów.
• Możesz chcieć stworzyć logikę po stronie klienta, która odrzuca zduplikowane wiadomości wysyłane z serwera, ponieważ jest to technicznie możliwe, jak niektórzy zauważyli.
• Następnie, gdy klient rozłącza się z serwerem gniazda (celowo lub przez błąd), NIE usuwaj wiersza klienta, po prostu wyczyść co najwyżej identyfikator gniazda. Dzieje się tak, ponieważ ten sam klient może spróbować połączyć się ponownie.
• Gdy klient próbuje ponownie się połączyć, wyślij ten sam identyfikator klienta, który wysłał przy pierwotnej próbie połączenia. Serwer będzie traktował to tak, jak początkowe połączenie.
• Gdy klient zostaje zniszczony (użytkownik zamyka kartę lub opuszcza ją), ma to miejsce podczas usuwania wiersza klienta i wszystkich wiadomości dla tego klienta. Ten krok może być nieco trudny.

Ponieważ ostatni krok jest trudny (przynajmniej był kiedyś, nie robiłem niczego takiego od dłuższego czasu) i ponieważ są przypadki takie jak utrata zasilania, w których klient rozłącza się bez czyszczenia wiersza klienta i nigdy nie próbuje aby ponownie połączyć się z tym samym wierszem klienta - prawdopodobnie chcesz mieć coś, co uruchamia się okresowo, aby wyczyścić przestarzałe wiersze klienta i wiadomości. Lub możesz po prostu trwale przechowywać wszystkich klientów i wiadomości na zawsze i po prostu odpowiednio oznaczyć ich stan.

Tak więc, żeby było jasne, w przypadkach, gdy jeden użytkownik ma otwarte dwie karty, dodasz dwie identyczne wiadomości do tabeli wiadomości, z których każda jest oznaczona dla innego klienta, ponieważ serwer musi wiedzieć, czy otrzymał je każdy klient, a nie tylko każdy użytkownik.

Wygląda na to, że masz już system kont użytkowników. Wiesz, które konto jest online / offline, możesz obsłużyć zdarzenie połączenia / rozłączenia:

Tak więc rozwiązaniem jest dodanie wiadomości online / offline i offline w bazie danych dla każdego użytkownika:

chatApp.onLogin(function (user) {
   user.readOfflineMessage(function (msgs) {
       user.sendOfflineMessage(msgs, function (err) {
           if (!err) user.clearOfflineMessage();
       });
   })
});

chatApp.onMessage(function (fromUser, toUser, msg) {
   if (user.isOnline()) {
      toUser.sendMessage(msg, function (err) {
          // alert CAN NOT SEND, RETRY?
      });
   } else {
      toUser.addToOfflineQueue(msg);
   }
})

Spójrz tutaj: Obsługa przeładowania przeglądarki socket.io .

Myślę, że przydałoby się rozwiązanie, które wymyśliłem. Jeśli odpowiednio go zmodyfikujesz, powinno działać tak, jak chcesz.

Myślę, że chcesz mieć gniazdo wielokrotnego użytku dla każdego użytkownika, na przykład:

Klient:

socket.on("msg", function(){
    socket.send("msg-conf");
});

Serwer:

// Add this socket property to all users, with your existing user system
user.socket = {
    messages:[],
    io:null
}
user.send = function(msg){ // Call this method to send a message
    if(this.socket.io){ // this.io will be set to null when dissconnected
        // Wait For Confirmation that message was sent.
        var hasconf = false;
        this.socket.io.on("msg-conf", function(data){
            // Expect the client to emit "msg-conf"
            hasconf = true;
        });
        // send the message
        this.socket.io.send("msg", msg); // if connected, call socket.io's send method
        setTimeout(function(){
            if(!hasconf){
                this.socket = null; // If the client did not respond, mark them as offline.
                this.socket.messages.push(msg); // Add it to the queue
            }
        }, 60 * 1000); // Make sure this is the same as your timeout.

    } else {
        this.socket.messages.push(msg); // Otherwise, it's offline. Add it to the message queue
    }
}
user.flush = function(){ // Call this when user comes back online
    for(var msg in this.socket.messages){ // For every message in the queue, send it.
        this.send(msg);
    }
}
// Make Sure this runs whenever the user gets logged in/comes online
user.onconnect = function(socket){
    this.socket.io = socket; // Set the socket.io socket
    this.flush(); // Send all messages that are waiting
}
// Make sure this is called when the user disconnects/logs out
user.disconnect = function(){
    self.socket.io = null; // Set the socket to null, so any messages are queued not send.
}

Następnie kolejka gniazda jest zachowywana między rozłączeniami.

Upewnij się, że zapisuje każdą socketwłaściwość użytkownika w bazie danych i uczyń metody częścią prototypu użytkownika. Baza danych nie ma znaczenia, po prostu zapisz ją, jakkolwiek zapisałeś swoich użytkowników.

Pozwoli to uniknąć problemu wspomnianego w Dodatku 1, wymagając potwierdzenia od klienta przed oznaczeniem wiadomości jako wysłanej. Jeśli naprawdę chcesz, możesz nadać każdej wiadomości identyfikator i poprosić klienta o wysłanie identyfikatora wiadomości msg-conf, a następnie sprawdzić to.

W tym przykładzie userjest to użytkownik szablonu, z którego kopiowani są wszyscy użytkownicy, lub podobny do prototypu użytkownika.

Uwaga: to nie zostało przetestowane.

Jak już napisano w innej odpowiedzi, uważam również, że jako bonus należy spojrzeć na czas rzeczywisty: system powinien być w stanie pracować również bez czasu rzeczywistego.

Rozwijam czat korporacyjny dla dużej firmy (iOS, Android, interfejs sieciowy i .net core + postGres) i po opracowaniu sposobu, w jaki websocket może ponownie nawiązać połączenie (przez UUID gniazda) i otrzymać niedostarczone wiadomości (przechowywane w kolejce) Zrozumiałem, że jest lepsze rozwiązanie: ponowna synchronizacja przez rest API.

Zasadniczo skończyło się na używaniu websocket tylko w czasie rzeczywistym, z tagiem typu integer w każdej wiadomości w czasie rzeczywistym (użytkownik online, piszący, wiadomość na czacie itd.) Do monitorowania utraconych wiadomości.

Gdy klient otrzyma identyfikator, który nie jest monolityczny (+1), wówczas rozumie, że nie jest zsynchronizowany, więc porzuca wszystkie komunikaty gniazda i prosi o ponowną synchronizację wszystkich swoich obserwatorów przez REST api.

W ten sposób możemy obsłużyć wiele wariacji stanu aplikacji w okresie offline bez konieczności analizowania ton wiadomości z WebSocket w rzędzie przy ponownym połączeniu i jesteśmy pewni, że zostaniemy zsynchronizowani (ponieważ data ostatniej synchronizacji jest ustawiana tylko przez REST api , nie z gniazda).

Jedyną trudną częścią jest monitorowanie wiadomości w czasie rzeczywistym od momentu wywołania REST API do momentu odpowiedzi serwera, ponieważ to, co jest odczytywane z bazy danych, wymaga czasu, aby wrócić do klienta, a w międzyczasie mogą wystąpić różnice, więc muszą być buforowane i wziął pod uwagę.

Za kilka miesięcy wchodzimy do produkcji, mam nadzieję, że do tego czasu znów będę spać :)

Patrzyłem na to ostatnio i myślę, że inna ścieżka może być lepsza.

Spróbuj spojrzeć na usługę Azure Service Bus, pytania i tematy, aby zająć się stanami offline. Wiadomość czeka, aż użytkownik wróci, a następnie otrzyma wiadomość.

Jest to koszt uruchomienia kolejki, ale jest to około 0,05 USD za milion operacji w przypadku podstawowej kolejki, więc koszt dewelopera byłby większy od godzin pracy potrzebnych do napisania systemu kolejkowego. https://azure.microsoft.com/en-us/pricing/details/service-bus/

A Azure Bus ma biblioteki i przykłady dla PHP, C #, Xarmin, Anjular, Java Script itp.

Serwer wysyła więc wiadomość i nie musi martwić się o ich śledzenie. Klient może użyć wiadomości do odesłania, a także jako sposób obsługi równoważenia obciążenia, jeśli to konieczne.

Wypróbuj tę listę wysyłanych czatów

io.on('connect', onConnect);

function onConnect(socket){

  // sending to the client
  socket.emit('hello', 'can you hear me?', 1, 2, 'abc');

  // sending to all clients except sender
  socket.broadcast.emit('broadcast', 'hello friends!');

  // sending to all clients in 'game' room except sender
  socket.to('game').emit('nice game', "let's play a game");

  // sending to all clients in 'game1' and/or in 'game2' room, except sender
  socket.to('game1').to('game2').emit('nice game', "let's play a game (too)");

  // sending to all clients in 'game' room, including sender
  io.in('game').emit('big-announcement', 'the game will start soon');

  // sending to all clients in namespace 'myNamespace', including sender
  io.of('myNamespace').emit('bigger-announcement', 'the tournament will start soon');

  // sending to individual socketid (private message)
  socket.to(<socketid>).emit('hey', 'I just met you');

  // sending with acknowledgement
  socket.emit('question', 'do you think so?', function (answer) {});

  // sending without compression
  socket.compress(false).emit('uncompressed', "that's rough");

  // sending a message that might be dropped if the client is not ready to receive messages
  socket.volatile.emit('maybe', 'do you really need it?');

  // sending to all clients on this node (when using multiple nodes)
  io.local.emit('hi', 'my lovely babies');

};