ついにここに...RocketMQ リテラシーの章
- coldplay.xixi転載
- 2020-10-20 17:16:392974ブラウズ
java Basic Tutorial コラムでは、今日の RocketMQ に関する知識を詳しく紹介します。
ブログを書くのは久しぶりです。ブログを書かない理由は数え切れないほどありますが、結局のところ、やはり「怠惰」です。今日はやっと怠け者のガンを治す薬を飲み、ブログを書くことにしました。何を紹介しようか? いろいろ考えた結果、RocketMQ を紹介することにします。結局、30 個以上のブログを書いてきましたが、まだ MQ についてまともなブログを書いたことはありません。このブログは比較的基本的なものであり、ソース コードの分析は含まれておらず、リテラシーだけが含まれています。
MQ の用途について
デカップリング
ある視点から見ると、マイクロサービスが MQ の活発な開発を促進したと思います。本来、システムには N 個の複数のモジュールがあり、すべてのモジュールは強く結合されています。マイクロサービスでは、モジュールはシステムであり、システムは間違いなく対話する必要があります。対話には 3 つの一般的な方法があります。1 つは RPC、1 つは HTTP、もう 1 つは MQ です。
非同期
当初、ビジネスは N のステップに分割されており、最終結果をユーザーに返す前に段階的に処理する必要がありましたが、MQ では最も重要な部分が実現されました。が最初に処理され、その後 MQ にメッセージを送信し、ユーザーに直接 OK を返します。以降の手順については、バックグラウンドでゆっくりと処理しましょう。これはユーザー エクスペリエンスを向上させるための成果物です。
ピーク クリッピング
特定のインターフェイスに対するリクエストの数が突然急増すると、必然的にアプリケーション サーバーとデータベース サーバーに多大な負荷がかかります。バックグラウンドでゆっくりと処理されるため、どれだけのリクエストが来るかを気にする必要はありません。
RocketMQ の概要
RocketMQ は Java で書かれており、Alibaba のオープン ソース メッセージ ミドルウェアであり、Kafka の利点の多くを吸収しています。 Kafka も人気のあるメッセージ ミドルウェアですが、Kafka は Scala で書かれているため、Java プログラマーがソース コードを読むのには適していません。また、Java プログラマーがカスタマイズされた開発を行うのにも適していません。 Kafka に触れたことのある友人は、Kafka を使いこなすのが簡単ではないことを知っています。比較的言えば、RocketMQ ははるかに単純であり、RocketMQ は Alibaba の恩恵を受けており、N Double 11 のテストを経験しています。国内のインターネット企業により適しています。なので国内でも使われており、RocketMQの会社はたくさんあります。
RocketMQ の 4 つの主要コンポーネント
画像は gitee.com/mirrors/roc からのものです...
RocketMQ には 4 つの主要コンポーネントがあることがわかります。 :
NameServer
- #ステートレス サービスである登録センターはクラスターに展開できますが、NameServer ノード間でのデータのやり取りはありません。
- Borker は、トピック ルーティング情報をすべてのネームサーバーに定期的に報告します。プロデューサーとコンシューマーは、ルーティング情報を定期的に更新するためにネームサーバー トピックをランダムに選択します。
- トピック ルーティング情報は、NameServer クラスター内の結果整合性を採用します。
- 保証されたAP。
- RocketMQ のサーバーは、メッセージの保存と配布に使用されます。
- Borker は、所有するすべてのトピック ルーティング情報を NameServer に定期的に報告します。
- ボーカーには、マスターとフォロワーの 2 つの役割があります。マスターは、読み取り (メッセージの消費) 操作と書き込み (メッセージの生成) 操作を担当します。マスターがビジーまたは応答できない場合、フォロワーは読み取り操作を引き受けることができます。 BorkerId=0 は Matser、BorkerId!=0 は Follower を意味します。 まず、これまでのところ、BorkerId=1 を持つフォロワーのみが読み取り操作を実行できます。 第 2 に、マスター ノードがハングアップした場合のフォロワーからマスターへの自動アップグレードをサポートしているのは、RocketMQ の上位バージョンのみです。
- RocketMQ はすでにミドルウェアであるため、他のミドルウェアに依存したくありません。
- Zookeeper は比較的重く、RocketMQ では使用できない機能が多いため、軽量な登録センターを作成することをお勧めします。
- Zookeeper は CP です。リーダー選挙がトリガーされると、登録センターは利用できなくなります。ただし、RocketMQ の登録センターは、結果整合性が確保されている限り、強い一貫性を必要としません。
RocketMQ メッセージ ドメイン モデル
Message
- 送信されたメッセージ。
- メッセージにはトピックが必要です。
- メッセージには複数のタグと複数のキーを含めることができ、これらはメッセージの追加属性とみなすことができます。
トピック
- 1 種類のメッセージのコレクション。
- 各メッセージにはトピックが必要です。
- 第 1 レベルのメッセージのタイプ。
タグ
- トピックに加えて、メッセージにはタグを付けることもできます。タグは、同じトピックの下でさまざまな種類のメッセージを細分化するために使用されます。
- タグは必要ありません。
- 第 2 レベルのメッセージのタイプ。
Group
ProducerGroup と ConsumerGroup に分かれていますが、ConsumerGroup に注目します。ConsumerGroup には複数の Consumer が含まれます。
クラスター消費モードでは、ConsumerGroup の下のコンシューマが一緒にトピックを消費し、各コンシューマは N 個のキューに割り当てられますが、キューは 1 つのコンシューマによってのみ消費されます。異なる ConsumerGroup は同じ A トピックを消費でき、メッセージは、このトピックにサブスクライブされているすべての ConsumerGroups によって消費されます。
キュー
- トピックには、デフォルトで 4 つのキューが含まれています。
- クラスター消費モードでは、同じ ConsumerGroup 内の Consumer は複数の Queue からのメッセージを消費できますが、1 つの Queue は 1 つの Consumer によってのみ消費されます。
- キュー内のメッセージは順序付けされます。
- 読み取りキューと書き込みキューに分かれており、一般に読み取りキューと書き込みキューの数が一致していないと問題が発生しやすくなります。
消費モード
クラスタリング (クラスター消費) とブロードキャスト (ブロードキャスト消費) の 2 つの消費モードがあります。
他の MQ とは異なり、他の MQ はメッセージ送信時にクラスター消費またはブロードキャスト消費を指定しますが、RocketMQ はクラスター消費またはブロードキャスト消費をコンシューマ側で設定します。
クラスタリング (クラスター消費)
デフォルトはクラスター消費モードです。このモードでは、ConsumerGroup のすべての Consumer が共同でトピックからのメッセージを消費し、各 Consumer は N からのメッセージを消費する責任があります。 (N は 1、またはキューに割り当てられていない 0 の場合もあります)、キューは 1 つのコンシューマによってのみ消費されます。 Consumer が死亡した場合、ConsumerGroup の下にある他の Consumer が引き継ぎ、消費を続けます。
クラスター消費モードでは、消費の進行状況はボーカー側で維持され、ストレージ パスは ${ROCKET_HOME}/store/config/consumerOffset.json です (次に示すように)。図: 
UsetopicName@consumerGroupName は Key、消費の進行状況は Value です。Value の形式は queueId:offset で、複数の ConsumerGroup がある場合、消費の進行状況が各 ConsumerGroup の進行状況は異なるため、分離する必要があります。
ブロードキャスト (ブロードキャスト消費)
ブロードキャスト消費メッセージは、ConsumerGroup 内のすべての Consumer に送信されます。
ブロードキャスト消費モードでは、消費の進行状況はコンシューマ側で維持されます。
消費キューのロード アルゴリズムとリバランス メカニズム
消費キューのロード アルゴリズム
クラスター消費モードでは、ConsumerGroup の下のすべての Consumer が共同で Topic メッセージを消費することがわかっています。 N 個のキューからのメッセージを消費する責任がありますが、どのように割り当てられるのでしょうか?これには、消費キューのロード アルゴリズムが関係します。
RocketMQ は多数のコンシューマー キュー ロード アルゴリズムを提供します。その中で最もよく使用される 2 つのアルゴリズムは、AllocateMessageQueueAveragely と AllocateMessageQueueAveragelyByCircle です。これら 2 つのアルゴリズムの違いを見てみましょう。
トピックには、q0 ～ q15 で表される 16 個のキューと、c0 ～ c2 で表される 3 つのコンシューマがあるとします。
AllocateMessageQueueAveragely を使用してキュー負荷アルゴリズムを使用した結果は次のとおりです。
- c0：q0 q1 q2 q3 q4 q5
- c1：q6 q7 q8 q9 q10
- c2：q11 q12 q13 q14 q15
AllocateMessageQueueAveragelyByCircle を使用してキュー負荷アルゴリズムを使用した結果は次のとおりです。
- c0：q0 q3 q6 q9 q12 q15
- c1: q1 q4 q7 q10 q13
- c2: q2 q5 q8 q11 q14
ConsumerGroup 下のすべての Consumer が一緒にトピック メッセージを消費します、各コンシューマは N 個のキュー メッセージを消費する責任がありますが、1 つのキューを N 個のコンシューマが同時に消費することはできません。これは何を意味しますか?
賢明な方なら、トピックにキューが 4 つとコンシューマが 5 つしかない場合、1 つのコンシューマはどのキューにも割り当てられないと考えたはずです。そのため、RocketMQ では、トピックの下のキューの数は次のようになります。コンシューマの最大数を直接決定します。つまり、コンシューマを追加するだけでは消費速度を上げることはできません。
リバランス
トピックを作成する際にはキューの数を十分に考慮することが推奨されますが、実際の状況は満足できない場合が多く、キューの数が変わらなくてもコンシューマの数は変化します。コンシューマがオンラインまたはオフラインになったとき、コンシューマが電話を切ったとき、または新しいコンシューマが追加されたときなどに、変化が発生します。キューの拡大と縮小、およびコンシューマの拡大と縮小により、リバランスが行われます。つまり、消費キューがコンシューマに再分配されます。
RocketMQ では、コンシューマーはトピック キューの数を定期的にクエリします。コンシューマーの数が変化すると、リバランスがトリガーされます。
リバランスは RocketMQ によって内部的に実装されるため、プログラマは気にする必要はありません。
引くか押すか?
一般的に、MQ にはメッセージを取得する 2 つの方法があります:
- プル: コンシューマが積極的にメッセージをプルします。利点は、コンシューマがプルするメッセージの頻度と数を制御できることです。自身の消費容量が大きいため、Consumer 側でメッセージの蓄積を引き起こすのは簡単ではありませんが、リアルタイム性はあまり良くなく、効率は比較的低くなります。
- Push: Broker は積極的にメッセージを送信するため、リアルタイム性と効率性が高いという利点があります。しかし、Broker は Consumer の消費容量を知ることができません。Consumer にあまりにも多くのメッセージが送信されると、蓄積が発生します。 Consumer 側のメッセージの数; if send to Consumer のデータが少なすぎる場合、Consumer 側はアイドル状態になります。
プルでもプッシュでも、コンシューマは常にブローカーと対話します。対話方法には通常、短い接続、長い接続、ポーリングが含まれます。
RocketMQ はプルとプッシュの両方をサポートしているように見えますが、実際にはプッシュもプルを使用して実装されています。
これは RocketMQ の設計の独創的な部分で、短い接続でも長い接続でもポーリングでもなく、長いポーリングです。
ロングポーリング
コンシューマはメッセージをプルするリクエストを開始します。これは 2 つの状況に分けられます:
- メッセージがあります: コンシューマがメッセージを取得した後、接続が切断されました。
- メッセージなし: ボーカーは一定時間接続を保持します。5 秒ごとにメッセージがあるかどうかを確認します。メッセージがある場合は、コンシューマーと接続に送信されます。切断されています。
トランザクション メッセージ
RocketMQ はトランザクション メッセージをサポートします。プロデューサーがトランザクション メッセージをブローカーに送信した後、ブローカーはシステム トピック: RMQ_SYS_TRANS_HALF_TOPIC## にメッセージを保存します。 #、Consumer This メッセージを消費できないようにします。
RMQ_SYS_TRANS_HALF_TOPIC メッセージを消費し、プロデューサーへのレビューを開始します。レビューには、送信済み、ロールバック、不明の 3 つのステータスがあります。
- レビューのステータスが送信またはロールバックの場合、メッセージの送信とロールバックがトリガーされます。
- #不明な場合は、次のレビューを待ちます。 RocketMQ では、メッセージのチェックバック間隔とチェックバック回数を設定でき、チェックバック回数が一定の回数を超えると、メッセージは自動的にロールバックされます。
1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。
- 同期フラッシュ: Borker は、Producer に ACK を返す前にメッセージを永続化するため、メッセージの信頼性が高いという利点がありますが、効率は遅くなります。
- 非同期フラッシュ: ブローカーはメッセージを PageCache に書き込み、プロデューサーに ACK を返します。利点は非常に効率的であることですが、サーバーがハングアップするとメッセージが失われる可能性がありますが、RocketMQ サービスのみがハングアップする場合はメッセージは失われません。
- 同期レプリケーション: マスターとフォロワーの両方が、プロデューサーに ACK を返す前にメッセージを正常に書き込みます 信頼性は高くなりますが、効率は遅くなります。
- 非同期レプリケーション: マスターが正常に書き込みを行う限り、ACK がプロデューサーに返されるため効率的ですが、メッセージが失われる可能性があります。
- oneway: ファイアアンドフォーゲット、一方向メッセージ、つまりメッセージが送信された後は無視されます。このメソッドには戻り値がありません。
- 同期: メッセージが送信された後、同期的にボーカーの応答を待ちます。
- 非同期: メッセージ送信後すぐに戻り、Boker の応答を受信した後、関数呼び出しメソッドが実行されます。
実際の開発では同期方式が一般的ですが、RocketMQ のパフォーマンスを向上させたい場合は、Borker 側のパラメータ、特にディスク ブラッシング戦略とレプリケーション戦略を変更することが一般的です。
メッセージ送信の再試行
メッセージ送信時に、MessageQueueSelector を使用すると、メッセージ送信の再試行機構が無効になります。
メッセージ送信に対する応答は、次の 4 つになる可能性があります:
public enum SendStatus {
SEND_OK,
FLUSH_DISK_TIMEOUT,
FLUSH_SLAVE_TIMEOUT,
SLAVE_NOT_AVAILABLE,
}复制代码
最初の状況を除き、他の状況は問題があります。メッセージが失われないようにするには、次のことが必要です。プロデューサー パラメーターを設定するには: RetryAnotherBrokerWhenNotStoreOK が true です。
障害回避メカニズム
メッセージの送信に失敗した場合でも、再試行時にメッセージがボーカーに送信されるため、送信が失敗する可能性が高くなります。 RockteMQ の特徴は再試行時間であり、自動的にこの Borker を回避し、他の Borker を選択しますが、これまでのところ、非同期送信はそれほどスマートではなく、1 つの Borker でのみ再試行されるため、同期送信方法を選択することを強くお勧めします。
RocketMQ は 2 つの障害回避メカニズムを提供します。パラメータ SendLatencyFaultEnable を使用して制御します。
- false: デフォルト値。障害回避メカニズムは再試行時にのみ有効になります。たとえば、BorkerA へのメッセージの送信が失敗した場合、再試行時には BorkerB が選択されますが、メッセージは次に送信されます。 、それでも BorkerA に送信することを選択します。
- true: 遅延バックオフ メカニズムをオンにします。BorkerA へのメッセージの送信に失敗すると、BorkerA は一定期間利用できなくなると悲観的に判断し、それ以上のメッセージは BorkerA に送信されなくなります。将来のある期間。
遅延バックオフ メカニズムは非常に便利であるように見えますが、一般的に言えば、Borker 側がビジー状態であるため、Borker が利用できなくなったり、ネットワークが利用できなくなったりします。遅延バックオフのメカニズムがオンになっている場合、当初利用可能だったボーカーが一定期間回避され、他のボーカーが忙しくなり、状況がさらに悪化する可能性があります。
Consumer について話しましょう
Consumer スレッドの考慮事項
Consumer には、消費できる並列度を表す 2 つのパラメーターがあります。つまり、ConsumeThreadMin、## です。 # ConsumeThreadMax、コンシューマ側に蓄積されたメッセージが比較的少ない場合、コンシューマ スレッドの数は ConsumeThreadMin であるようです。コンシューマ側に蓄積されたメッセージがさらに多い場合、新しいスレッドが作成されます。コンシューマ スレッドの数が ConsumeThreadMax になるまで、消費のために自動的に開かれます。 Consumer は内部でスレッド プールを保持し、無制限のキューを使用します。つまり、ConsumeThreadMax パラメータは無効であるため、実際の開発では、ConsumeThreadMin, ConsumeThreadMax は、多くの場合、同じ値に設定されます。
%RETRY% ConsumerGroup というトピック名で再試行キューを設定し、送信する必要がある再試行キューを保存します。 ConsumerGroup に送信されます。再試行メッセージですが、再試行には一定の遅延時間が必要です。RocketMQ は、最初にトピック名 SCHEDULE_TOPIC_XXXX で遅延キューに再試行メッセージを保存することによって再試行メッセージを処理し、次にバックグラウンドでスケジュールされたタスクが以下に従って遅延されます。 %RETRY% ConsumerGroup の再試行キューに再保存します。
- これは、消費の進行状況を改善するための最良の方法です。
- キューを追加し、コンシューマを追加します。
- 元のコンシューマは、ブリック ムーバーとして、特定のルールに従ってメッセージを複数の新しいトピックに「移動」し、複数の ConsumerGroup を開いてさまざまなトピックを消費します。
- 消費のために新しい ConsumerGroup を開きます。つまり、2 つの ConsumerGroup が同時に Topic を消費しますが、オフセットの判断に注意する必要があります。たとえば、ConsumerGroup は奇数のメッセージを消費します。 ConsumerGroup は偶数のメッセージを消費します。
関連する無料学習の推奨事項: Java 基本チュートリアル
以上がついにここに...RocketMQ リテラシーの章の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。