Rabbitmq heartbea ハートビート検出メカニズムの原理の分析

輸入ナキウサギ 
インポート時間 

クレジット = pika.PlainCredentials(ユーザー名 = 'cloud'、パスワード = 'cloud')
接続 = pika.BlockingConnection(pika.ConnectionParameters(
  ホスト='10.32.1.12'、資格情報=credit))
チャネル = connection.channel() 
真の場合:
  connect_close = 接続が閉じられました
  connect_open = 接続が開いている
  チャネルが閉じている = チャネルが閉じている
  チャンネルオープン = チャンネルがオープンしている
  
  print("接続が閉じられました ", connect_close)
  print("接続はis_open"、connect_open)
  print("チャンネルは閉じています ", channel_close)
  print("チャンネルは開いている ", channel_open)
  印刷("")
  時間.睡眠(5)

輸入ナキウサギ

真の場合:
  試す：
    接続 = pika.BlockingConnection()
    チャネル = connection.channel()
    チャネル.basic_consume('テスト'、on_message_callback)
    チャネルの開始_消費()
  # ブローカーによって接続が閉じられた場合は回復しない
  pika.exceptions.ConnectionClosedByBroker を除く:
    壊す
  # チャネルエラー時に回復しない
  pika.exceptions.AMQPChannelError を除く:
    壊す
  # その他の接続エラーが発生した場合は回復する
  pika.exceptions.AMQPConnectionError を除く:
    続く

再試行からインポート再試行

@retry(pika.exceptions.AMQPConnectionError、遅延=5、ジッター=(1、3))
def 消費():
  接続 = pika.BlockingConnection()
  チャネル = connection.channel()
  チャネル.basic_consume('テスト'、on_message_callback)
  試す：
    チャネルの開始_消費()
  # サーバーによって閉じられた接続を回復しない
  pika.exceptions.ConnectionClosedByBroker を除く:
    合格
スタブ（）

開始(SupPid、Sock、SendTimeoutSec、
   SendFun、ReceiveTimeoutSec、ReceiveFun) ->
  %%データ送信検出プロセス {ok, Sender} = start_heartbeater(SendTimeoutSec, SupPid, Sock,
                   SendFun、heartbeat_sender、
                   start_heartbeat_sender)、
  %%データ受信検出プロセス {ok, Receiver} = start_heartbeater(ReceiveTimeoutSec, SupPid,
                    ソックス、レシーブファン、
                    ハートビートレシーバー、
                    start_heartbeat_receiver)、
  {送信者、受信者}。

start_heartbeat_sender(Sock、TimeoutSec、SendFun) ->
  %% 'div 2' は、待つ必要がないようにするためにあります
  %% ハートビートを送信する前に約2 * TimeoutSec
  %% 境界ケース
  ハートビーター({Sock, TimeoutSec * 1000 div 2, send_oct, 0,
         fun () -> SendFun()、続行 end})。

start_heartbeat_receiver(Sock、TimeoutSec、ReceiveFun) ->
  %% 一定間隔ごとに受信データをチェックし、一定時間後にタイムアウトします
  %% 2回のチェックで変化なし。結果としてタイムアウトになります
  %% 最後のデータが取得されてから2～3間隔の間
  ％％ 受け取った
  ハートビーター({Sock, TimeoutSec * 1000, recv_oct, 1,
        fun () -> ReceiveFun()、stop end})。

ハートビーター({Sock, TimeoutMillisec, 
       StatName、しきい値、ハンドラー} = パラメータ、
      デブ、
      {StatVal, SameCount} = 状態) ->
  再帰 = fun (State1) -> heartbeater(Params, Deb, State1) 終了、
  受け取る
    ...
  %% TimeoutMillisec 後の時間検出 ->
    ケースrabbit_net:getstat(Sock, [StatName]) の
      {ok、[{StatName, NewStatVal}]} ->
        %% 送受信されたデータは、NewStatVal =/= StatVal の場合に変更されました ->
            %%検出を再開する Recurse({NewStatVal, 0});
          %% 指定回数に達しなかった場合は送信は0、受信は1となります
          同じカウント<しきい値->
            %%Count に 1 を加算して再度チェックしますRecurse({NewStatVal, SameCount + 1});
          %%ハートビートタイムアウト true ->
            %% 送信検出タイムアウトの場合、クライアントにハートビートパケットを送信します %% 受信検出タイムアウトの場合、親プロセスにタイムアウト通知を送信します %% 親プロセスはTCPシャットダウンおよびその他の操作をトリガーします case Handler() の
              %%受信検出タイムアウト停止 -> ok;
              %%送信検出タイムアウト継続 -> Recurse({NewStatVal, 0})
            終わり;
      ...