Analyse des Prinzips des Rabbitmq Heartbea-Herzschlagerkennungsmechanismus

Vorwort

Wenn Sie RabbitMQ verwenden und für einen bestimmten Zeitraum kein Datenverkehr zwischen Ihrem Client und dem RabbitMQ-Server stattfindet, trennt der Server die TCP-Verbindung mit dem Client.

Und Sie werden Protokolle wie diese auf dem Server sehen:

Verpasste Heartbeats vom Client, Timeout: xxs

Dieses Intervall ist das Heartbeat-Intervall.

Heartbeat wird normalerweise verwendet, um zu erkennen, ob das andere Ende der Kommunikation aktiv ist (abnormaler Absturz aufgrund eines Fehlers beim normalen Schließen der Socket-Verbindung). Das Grundprinzip besteht darin, festzustellen, ob die Datenübertragung und der Datenübertragungsempfang über die entsprechende Socket-Verbindung normal sind. Wenn für einen bestimmten Zeitraum keine Daten gesendet oder empfangen werden, wird ein Heartbeat-Erkennungspaket an das andere Ende gesendet. Wenn innerhalb eines bestimmten Zeitraums keine Antwort erfolgt, wird davon ausgegangen, dass der Heartbeat abgelaufen ist, d. h., es wird davon ausgegangen, dass das andere Ende möglicherweise abnormal abgestürzt ist.

RabbitMQ ist keine Ausnahme. Heatbeat wird zwischen dem Client und dem Server verwendet, um festzustellen, ob das andere Ende normal ist, d. h. ob die TCP-Verbindung zwischen dem Client und dem Server normal ist.

Über RabbitMQ Heartbeat

1. Das Zeitintervall für die Heartbeat-Erkennung kann durch Hinzufügen des Konfigurationselements {heartbeat,Timeout} in der Konfigurationsdatei rabbitmq.config konfiguriert werden, wobei Timeout das Zeitintervall in Sekunden angibt. Darüber hinaus kann der Client auch die Heartbeat-Zeit konfigurieren.

Wenn der Server nicht konfiguriert ist

Standardmäßige Proxy-Heartbeat-Zeit:

RabbitMQ 3.2.2: 580 Sekunden
RabbitMQ 3.5.5: 60 Sekunden

2. Die offizielle Empfehlung lautet, den Heartbeat nicht zu deaktivieren. Die empfohlene Heartbeat-Zeit beträgt 60 Sekunden.

3. Der Heartbeat wird alle 2 Sekunden gesendet, wenn der Heartbeat-Timeout abgelaufen ist. Wenn der Server ihn nicht zweimal empfängt, wird die TCP-Verbindung getrennt, die vorherige Verbindung ist ungültig und der Client muss erneut eine Verbindung herstellen.

4. Wenn Sie Java-, .NET- und Erlang-Clients verwenden, verhandeln Server und Client die Heartbeat-Zeit

Wenn einer der Werte 0 ist, wird der größere der beiden verwendet.

Andernfalls verwenden Sie den kleineren der beiden

Wenn beide Werte 0 sind, bedeutet dies, dass der Heartbeat deaktiviert ist und der Server und der Client diese TCP-Verbindung aufrechterhalten und nicht trennen.

Hinweis: Wenn Sie diesen Wert direkt auf dem Python-Client auf 0 setzen, wird der Heartbeat deaktiviert.

So deaktivieren Sie Heartbeat im Python-Client:

Setzen Sie einfach heartbeat_interval=0 in py3:ConnectionParameters.

Setzen Sie einfach heartbeat=0 in py2:ConnectionParameters.

5. Jeglicher Datenverkehr auf der Verbindung (übertragene gültige Daten, Bestätigungen usw.) wird als gültiger Heartbeat gezählt, einschließlich Heartbeat-Frames.

6. Ich habe gesehen, wie jemand online diese Frage gestellt hat:

Warum stürzt der Server ab? Beim Heartbeat-Erkennungsmechanismus wird die Serverseite getrennt, aber die Clientseite kann die TCP-Trennung nicht erkennen. Ich habe getestet, dass der Client die Trennung der TCP-Verbindung nicht erkennen kann. Sie kann nur erkannt werden, wenn der Client eine Operation auf diesem TCP durchführt. Natürlich führt das Ausführen von Operationen auf einer getrennten TCP-Verbindung zu einem Fehler (z. B. das Senden einer Nachricht).

Pika importieren 
Importzeit 

Guthaben = pika.PlainCredentials(Benutzername='Cloud', Passwort='Cloud')
Verbindung = pika.BlockingConnection(pika.ConnectionParameters(
  Host = '10.32.1.12', Anmeldeinformationen = Kredit))
Kanal = Verbindung.Kanal() 
während True:
  connect_close = Verbindung ist geschlossen
  connect_open = Verbindung ist offen
  channel_close = Kanal ist geschlossen
  channel_open = Kanal.ist_offen
  
  drucken("Verbindung ist geschlossen ", connect_close)
  drucken("Verbindung ist_offen ", Verbindung_offen)
  drucken("Kanal ist geschlossen ", Kanal schließen)
  drucken("Kanal ist_offen ", Kanal_offen)
  drucken("")
  Zeit.Schlaf(5)

7. Einige RabbitMQ-Clients (Bunny, Java, .NET, Objective-C, Swift) bieten einen Mechanismus zur automatischen Wiederherstellung von Verbindungen nach Netzwerkausfällen, während pika nur Verbindungsanomalien erkennen und dann die Verbindungen wiederherstellen kann.

Beispielcode: Erstellen Sie die Verbindung neu, indem Sie Verbindungsausnahmen erkennen:

Pika importieren

während True:
  versuchen:
    Verbindung = pika.BlockingConnection()
    Kanal = Verbindung.Kanal()
    channel.basic_consume('test', bei_Nachrichtenrückruf)
    Kanal.start_using()
  # Nicht wiederherstellen, wenn die Verbindung vom Broker geschlossen wurde
  außer pika.exceptions.ConnectionClosedByBroker:
    brechen
  # Bei Kanalfehlern keine Wiederherstellung durchführen
  außer pika.exceptions.AMQPChannelError:
    brechen
  # Wiederherstellung bei allen anderen Verbindungsfehlern
  außer pika.exceptions.AMQPConnectionError:
    weitermachen

Sie können auch eine Bibliothek zur Wiederholung von Vorgängen verwenden, z. B. „retry“.

von Wiederholungsimport Wiederholungs

@retry(pika.exceptions.AMQPConnectionError, Verzögerung=5, Jitter=(1, 3))
def verbrauchen():
  Verbindung = pika.BlockingConnection()
  Kanal = Verbindung.Kanal()
  channel.basic_consume('test', bei_Nachrichtenrückruf)
  versuchen:
    Kanal.start_using()
  # Vom Server geschlossene Verbindungen nicht wiederherstellen
  außer pika.exceptions.ConnectionClosedByBroker:
    passieren
stub()

Implementierung von Heartbeat

Nach dem Empfang der connection.tune-ok-Signalisierung vom Client aktiviert rabbitmq die Heartbeat-Erkennung. rabbitmq erstellt für jede TCP-Verbindung zwei Prozesse zur Heartbeat-Erkennung. Ein Prozess erkennt regelmäßig, ob über die TCP-Verbindung Daten gesendet werden (das Senden bedeutet hier, dass rabbitmq Daten an den Client sendet). Wenn für einen bestimmten Zeitraum keine Daten an den Client gesendet werden, wird ein Heartbeat-Paket an den Client gesendet und anschließend die nächste Erkennung in einer Schleife ausgeführt; der andere Prozess erkennt regelmäßig, ob über die TCP-Verbindung Daten empfangen werden. Wenn für einen bestimmten Zeitraum keine Daten empfangen werden, wird festgestellt, dass der Heartbeat abgelaufen ist und die TCP-Verbindung schließlich geschlossen wird. Darüber hinaus kann der Flusskontrollmechanismus von RabbitMQ die Heartbeat-Erkennung aussetzen, was hier nicht beschrieben wird.

Beteiligter Quellcode:

start(SupPid, Sock, SendTimeoutSec,
   SendFun, ReceiveTimeoutSec, ReceiveFun) ->
  %%Datensende-Erkennungsprozess {ok, Sender} = start_heartbeater(SendTimeoutSec, SupPid, Sock,
                   SendFun, heartbeat_sender,
                   start_heartbeat_sender),
  %%Datenempfangserkennungsprozess {ok, Receiver} = start_heartbeater(ReceiveTimeoutSec, SupPid,
                    Socke, ReceiveFun,
                    Herzschlagempfänger,
                    start_heartbeat_receiver),
  {Absender, Empfänger}.

start_heartbeat_sender(Sock, TimeoutSec, SendFun) ->
  %% die 'div 2' ist da, damit wir nicht warten müssen auf
  %% fast 2 * TimeoutSec vor dem Senden eines Heartbeats im
  %% Grenzfall
  heartbeater({Sock, TimeoutSec * 1000 div 2, send_oct, 0,
         fun () -> SendFun(), weiter Ende}).

start_heartbeat_receiver(Socke, TimeoutSec, Empfangsspaß) ->
  %% wir prüfen in jedem Intervall auf eingehende Daten und es kommt zu einer Zeitüberschreitung nach
  %% zwei Überprüfungen ohne Änderung. Als Ergebnis werden wir eine Zeitüberschreitung haben
  %% zwischen 2 und 3 Intervallen nach der letzten Datenerfassung
  %% erhalten
  heartbeater({Sock, TimeoutSec * 1000, recv_oct, 1,
        fun () -> ReceiveFun(), Stopp Ende}).

Herzschlag({Sock, TimeoutMillisek, 
       StatName, Schwellenwert, Handler} = Params,
      Deb,
      {StatVal, SameCount} = Status) ->
  Rekursiv = Spaß (Zustand1) -> Heartbeater (Params, Deb, Zustand1) Ende,
  erhalten
    ...
  %% Zeitliche Erkennung nach TimeoutMillisekunden ->
    Fall rabbit_net:getstat(Sock, [StatName]) von
      {ok, [{StatName, NewStatVal}]} ->
        %% Die gesendeten und empfangenen Daten haben sich geändert, wenn NewStatVal =/= StatVal ->
            %%Neustarterkennung Recurse({NewStatVal, 0});
          %% Wenn die angegebene Anzahl nicht erreicht wird, ist das Senden 0 und das Empfangen 1
          SameCount<Schwellenwert->
            %%Count plus 1, nochmal prüfenRecurse({NewStatVal, SameCount + 1});
          %%Heartbeat-Timeout wahr ->
            %% Zum Senden einer Timeout-Erkennung wird ein Heartbeat-Paket an den Client gesendet. %% Zum Empfangen einer Timeout-Erkennung wird eine Timeout-Benachrichtigung an den übergeordneten Prozess gesendet. %% Der übergeordnete Prozess löst TCP-Shutdown und andere Operationen aus. case Handler() von
              %%Empfangserkennungs-Timeout, Stopp -> ok;
              %%Sendeerkennungs-Timeout, weiter -> Recurse({NewStatVal, 0})
            Ende;
      ...

Beim Senden und Empfangen wird getstat des Inet-Moduls verwendet, um die statistischen Informationen des Sockets anzuzeigen.

recv_oct: Zeigt die Anzahl der im Socket empfangenen Bytes an

send_oct: Zeigt die Anzahl der über den Socket gesendeten Bytes an.

Einzelheiten zu inet finden Sie hier: http://www.erlang.org/doc/man/inet.html

Das Obige ist der vollständige Inhalt dieses Artikels. Ich hoffe, er wird für jedermanns Studium hilfreich sein. Ich hoffe auch, dass jeder 123WORDPRESS.COM unterstützen wird.