Pub-Subシステムからのメッセージを処理する方法は多数あります：個別のサービスの使用、分離されたプロセスの分離、プロセス/スレッドプールの調整、複雑なIPC、Poll-over-Httpなど。 今日は、Pub-Sub over HTTPの使用方法と、このために特別に作成された独自のサービスについてお話したいと思います。

既製のHTTPサービスバックエンドを使用することは、メッセージキューを処理するための理想的なソリューションです。

1. 箱から出してバランスを取る。 通常、バックエンドはすでにバランサーの背後にあり、すぐにロードできるインフラストラクチャを備えているため、メッセージの処理が大幅に簡素化されます。
2. 通常のRESTコントローラー（任意のHTTPリソース）を使用します。 HTTPメッセージを使用すると、バックエンドが混在している場合に異なる言語のコンピューターを実装するコストが最小限に抑えられます。
3. 他のサービスのWebフックの使用の簡素化。 現在、ほぼすべてのサービス（Jira、Gitlab、Mattermost、Slackなど）が何らかの方法で外界とやり取りするためのWebフックをサポートしています。 HTTPディスパッチャーの機能を実行するようにキューを教えると、生活が楽になります。

このアプローチには欠点もあります。

1. ソリューションの軽さを忘れることができます。 HTTPは重いプロトコルであり、コンシューマ側でフレームワークを使用すると、遅延と負荷が即座に増加します。
2. 世論調査アプローチの長所を失い、プッシュの弱点を取得します。
3. クライアントを処理するのと同じサービスインスタンスでメッセージを処理すると、応答性に影響する場合があります。 バランスと分離で処理されるため、これは重要ではありません。

このアイデアをQueue-Over-Httpサービスとして実装しましたが、これについては後で説明します。 プロジェクトは、Spring Boot 2.1を使用してKotlinで記述されています。 ブローカーとして、現在利用できるのはApache Kafkaのみです。

さらにこの記事では、読者はKafkaに精通しており、メッセージのコミット（コミット）およびオフセット（オフセット）、グループの原則（グループ）およびコンシューマー（消費者）を知っていること、およびパーティション（パーティション）とトピック（トピック）の違いを理解していることを前提としています。 ギャップがある場合は、続行する前にKafkaのドキュメントのこのセクションを読むことをお勧めします。

内容

• 復習
• コミット
• エラー処理
• メッセージ
• 性能
• デモンストレーション
• おわりに

復習

Queue-Over-Httpは、メッセージブローカーと最終的なHTTPコンシューマーとの間の仲介役として機能するサービスです（このサービスにより、異なる* RPCなど、他の方法で消費者にメッセージを送信するためのサポートを簡単に実装できます）。 現時点では、コンシューマーのリストのサブスクリプション、アンサブスクライブ、および表示のみが使用可能です。プロデューサーからの特別なサポートなしではメッセージの順序を保証できないため、HTTPを介したブローカー（プロデュース）へのメッセージ送信はまだ実装されていません。

サービスの重要な人物は消費者であり、特定のパーティションまたはトピックのみをサブスクライブできます（トピックパターンがサポートされています）。 最初のケースでは、パーティションの自動バランスがオフになります。 サブスクライブ後、指定されたHTTPリソースは割り当てられたKafkaパーティションからメッセージを受信し始めます。 アーキテクチャ的には、各サブスクライバーはネイティブのKafka Javaクライアントに関連付けられています。


各パーティションは、アダプターがブローカーからのメッセージを書き込む個別のローカルキューで表されます。 ワーカーはそこからメッセージを取得し、実行のために、つまりHTTP経由で送信するために送信します。

このサービスは、スループットを向上させるバッチメッセージ処理をサポートしています。 サブスクライブするときにconcurrencyFactor各トピックのconcurrencyFactor指定できます（割り当てられた各パーティションに個別に適用されます）。 たとえば、 concurrencyFactor=1000は、HTTP要求の形式の1000のメッセージを同時にコンシューマに送信できることを意味します。 パックからのすべてのメッセージが消費者によって明確に解決されるとすぐに、サービスはKafkaの最後のメッセージのオフセットの次のコミットについて決定を下します。 したがって、 concurrencyFactorの2番目の値は、KafkaまたはQueue-Over-Httpクラッシュのイベントでコンシューマーによって処理されるメッセージの最大数です。

遅延を減らすために、キューにはloadFactor = concurrencyFactor * 2があります。これにより、ブローカーから送信できるメッセージの2倍のメッセージを読み取ることができます。 ネイティブクライアントでの自動コミットは無効になっているため、このスキームはAt-Least-Onceの保証に違反しません。
concurrencyFactor値を高くすると、最悪の場合に最大10ミリ秒かかるコミットの数が減るため、キューのスループットが向上します。 同時に、消費者の負荷が増加します。

バンドル内でメッセージを送信する順序は保証されていませんが、 concurrencyFactor=1設定することで実現できます。

コミット

コミットはサービスの重要な部分です。 データの次のパケットの準備ができると、パケットからの最後のメッセージのオフセットがすぐにKafkaにコミットされ、コミットが成功した後にのみ次のパケットが処理可能になります。 多くの場合、これでは不十分であり、自動コミットが必要です。 これを行うには、 autoCommitPeriodMsパラメーターがあります。これは、パーティションから読み取られた最後のメッセージをコミットするネイティブクライアントの従来の自動コミット期間とはほとんど関係ありません。 concurrencyFactor=10想像concurrencyFactor=10ください。 サービスは10個すべてのメッセージを送信し、それぞれの準備が整うまで待機しています。 メッセージ3の処理が最初に完了し、次にメッセージ1、次にメッセージ10が完了します。この時点で、自動コミットの時間に達しました。 At-Least-Onceセマンティクスに違反しないことが重要です。 したがって、その時点で正常に処理されたのは最初のメッセージ、つまりオフセット2のみであるため、コミットできます。 さらに、次の自動コミットまで、メッセージ2、5、6、4、および8が処理されますが、オフセット7のみをコミットする必要があります。 自動コミットはスループットにほとんど影響しません。

エラー処理

通常の操作モードでは、サービスはスーパーバイザにメッセージを1回送信します。 何らかの理由で4xxまたは5xxエラーが発生した場合、サービスはメッセージを再送信し、正常な処理を待機します。 試行間の時間は、個別のパラメーターとして構成できます。

また、メッセージが処理済みとしてマークされるまでの試行回数を設定することもできます。これにより、応答のステータスに関係なく再送信が停止されます。 機密データにこれを使用することはお勧めしません。消費者の失敗の状況は常に手動で調整する必要があります。 スティッキーメッセージは、サービスログと消費者の応答の状態を監視することで追跡できます。

メッセージ

ブローカーから抽出された各メッセージは、サブスクリプション中に指定されたリソースにHTTP経由でアドバイザーに送信されます。 デフォルトでは、メッセージは本文のPOSTリクエストによって送信されます。 この動作は、他の方法を指定することで変更できます。 メソッドが本文でのデータ送信をサポートしていない場合、メッセージが送信される文字列パラメーターの名前を指定できます。 さらに、サブスクライブするときに、各メッセージに追加される追加のヘッダーを指定できます。これは、トークンを使用した基本的な承認に便利です。 消費者、トピック、パーティションの識別子、メッセージ番号、メッセージ番号、パーティションキー（該当する場合）、およびブローカーの名前とともに、ヘッダーが各メッセージに追加されます。

性能

パフォーマンスを評価するために、サービスを実行するPC（Windows 10、OpenJDK-11（チューニングなしのG1）、i7-6700K、16GB）と、メッセージプロデューサー、HTTPが実行されるラップトップ（Windows 10、i5-8250U、8GB）デフォルト設定のリソースコンシューマとKafka。 PCは1Gb / sの有線接続を介してルーターに接続され、ラップトップは802.11acを介して接続されます。 100ミリ秒ごとに、110秒間のプロデューサーは、110バイトのメッセージを、異なるグループからコンシューマーがサブスクライブされている（ concurrencyFactor=500 、自動コミットがオフになっている）指定されたトピックに書き込みます。 スタンドは理想とはほど遠いですが、写真を撮ることができます。

重要な測定パラメータは、サービスの待ち時間への影響です。

させてください：
-t _q-ネイティブクライアントからメッセージを受信するサービスのタイムスタンプ
-d _t0は、t _qとメッセージがローカルキューからエグゼクティブのプールに送信された時間の間の時間です。
-d _tは、t _qとHTTP要求が送信された時間の間の時間です。 そのd _tは、メッセージのレイテンシに対するサービスの影響です。

測定中に、次の結果が得られました（C-消費者、T-トピック、M-メッセージ）：



標準動作モードでは、サービス自体はレイテンシにほとんど影響せず、メモリ消費は最小限です。 d _tの最大値（約60ミリ秒）は、サービス自体ではなくGCの動作に依存するため、特に示されていません。 GCの特別な調整またはG1をシェナンドーで置き換えることにより、最大値の広がりをスムーズにすることができます。

消費者がキューからのメッセージの流れに対処せず、サービスが調整モードをオンにすると、すべてが劇的に変化します。 このモードでは、要求に対する応答時間が大幅に増加するため、メモリ消費が増加し、リソースのタイムリーなクリーニングが妨げられます。 ここでの待機時間への影響は以前の結果と同じレベルのままであり、ローカルキューにメッセージをプリロードすることで高いdt値が発生します。

残念ながら、ラップトップはすでに1300 RPSで曲がっているので、より高い負荷でテストすることはできません。 誰かが高負荷での測定の組織を手伝うことができるなら、私は喜んでテスト用のアセンブリを提供します。

デモンストレーション

それではデモに移りましょう。 これには次のものが必要です。

• Kafkaブローカー、準備完了。 Bitnamiから192.168.99.100:9092で発生したインスタンスを取得します。
• メッセージを受信するHTTPリソース。 わかりやすくするために、SlackからWebフックを取りました。

まず、Queue-Over-Httpサービス自体を上げる必要があります。 これを行うには、空のapplication.ymlディレクトリに次のコンテンツを作成します。

 spring: profiles: default logging: level: com: viirrtus: queueOverHttp: DEBUG app: persistence: file: storageDirectory: "persist" brokers: - name: "Kafka" origin: "kafka" config: bootstrap.servers: "192.168.99.100:9092" 

ここでは、特定のブローカーの接続パラメーターと、起動時に失われないようにサブスクライバーを保存する場所をサービスに示します。 `app.brokers []。Config`では、ネイティブのKafkaクライアントでサポートされている接続パラメーターを指定できます;完全なリストはこちらにあります 。

構成ファイルはSpringによって処理されるため、そこに多くの興味深いことを書くことができます。 含めて、ロギングを構成します。

次に、サービス自体を実行します。 最も簡単な方法docker-compose.ymlを使用します。

 version: "2" services: app: image: viirrtus/queue-over-http:0.1.3 restart: unless-stopped command: --debug ports: - "8080:8080" volumes: - ./application.yml:/application.yml - ./persist:/persist 

このオプションが適切でない場合は、ソースからサービスをコンパイルできます。 プロジェクトのReadmeにあるアセンブリ手順。リンクは記事の最後にあります。

次のステップは、最初のサブスクライバーの登録です。 これを行うには、Consumerの説明とともにサービスへのHTTPリクエストを実行する必要があります。

 POST localhost:8080/broker/subscription Content-Type: application/json { "id": "my-first-consumer", "group": { "id": "consumers" }, "broker": "Kafka", "topics": [ { "name": "slack.test", "config": { "concurrencyFactor": 10, "autoCommitPeriodMs": 100 } } ], "subscriptionMethod": { "type": "http", "delayOnErrorMs": 1000, "retryBeforeCommit": 10, "uri": "<slack-wh-uri>", "additionalHeaders": { "Content-Type": "application/json" } } } 

すべてがうまくいけば、応答は送信されたコンテンツとほぼ同じになります。

各パラメーターを見てみましょう。

• Consumer.idサブスクライバーのID
• Consumer.group.idグループ識別子
• Consumer.broker購読する必要があるサービスブローカーを示します
• Consumer.topics[0].nameメッセージを受信するトピックの名前
• Consumer.topics[0].config. concurrencyFactor Consumer.topics[0].config. concurrencyFactor同時に送信されるメッセージの最大数
• Consumer.topics[0].config. autoCommitPeriodMs Consumer.topics[0].config. autoCommitPeriodMs準備完了メッセージの強制コミット期間
• Consumer.subscriptionMethod.typeサブスクリプションのタイプ。 現在、HTTPのみが使用可能です。
• Consumer.subscriptionMethod.delayOnErrorMsエラーで終了したメッセージを再送するまでの時間
• Consumer.subscriptionMethod.retryBeforeCommitエラーメッセージの再送信の試行回数。 0の場合-メッセージは処理が成功するまでスピンします。 私たちの場合、完全な配信の保証は、フローの一定性ほど重要ではありません。
• Consumer.subscriptionMethod.uriメッセージの送信先のリソース
• Consumer.subscriptionMethod.additionalHeader各メッセージとともに送信される追加ヘッダー。 Slackがリクエストを正しく解釈できるように、各メッセージの本文にJSONがあることに注意してください。

このリクエストでは、HTTPメソッドは省略されています。これは、デフォルトのPOSTであるSlackで十分だからです。

この時点から、サービスはslack.testトピックの割り当てられたパーティションの新しいメッセージを監視します。

トピックにメッセージを書き込むに/opt/bitnami/kafka/bin起動されたKafkaイメージの/opt/bitnami/kafka/binあるKafkaの組み込みユーティリティを使用します（他のKafkaインスタンスのユーティリティの場所は異なる場合があります）。

 kafka-console-producer.sh --broker-list localhost:9092 --topic slack.test > {“text”: “Hello!”} 

同時に、Slackは新しいメッセージを通知します：



コンシューマーのサブスクリプションを解除するには、サブスクリプション中と同じコンテンツを使用してPOSTリクエストを「ブローカー/サブスクリプション解除」するだけで十分です。

おわりに

現時点では、基本的な機能のみが実装されています。 さらに、バッチ処理を改善し、1回限りのセマンティクスを実装し、HTTP経由でブローカーにメッセージを送信する機能を追加し、最も重要なこととして、他の一般的なPub-Subのサポートを追加する予定です。

Queue-Over-Httpサービスは現在活発に開発中です。 バージョン0.1.3は、開発およびステージスタンドでのテストに十分安定しています。 パフォーマンスは、Windows 10、Debian 9、およびUbuntu 18.04でテストされています。 prodはご自身の責任で使用できます。 開発の支援やサービスに関するフィードバックをお望みの場合は、 Githubプロジェクトへようこそ。