SpringBootは分散メッセージングプラットフォームPulsarを統合

みなさんこんにちは。ジュン兄です。

優れたメッセージストリーミングプラットフォームとして、Pulsar はますます使用されています。この記事では、Pulsar の Java クライアントについて説明します。

パルサーを展開する

Pulsar をデプロイするには、ローカルバイナリインストール、Docker デプロイ、Kubernetes へのデプロイという 3 つの主な方法があります。

この記事では、docker を使用して単一ノードの Pulsar クラスターをデプロイします。実験環境は2コアCPU、4Gメモリです。

デプロイメントコマンドは次のとおりです。

 docker run -it -p 6650:6650 -p 8080:8080 --mount source=pulsardata,target=/pulsar/data --mount source=pulsarconf,target=/pulsar/conf apachepulsar/pulsar:2.9.1 bin/pulsar スタンドアロン

インストールプロセス中に次のエラーが発生する可能性があります。

不明なフラグ: --mount  
 「docker run --help」を参照してください。

これは、docker のバージョンが低く、マウントパラメータをサポートしていないためです。 docker バージョンを 17.06 以上にアップグレードするだけです。

ネットワーク上の理由により、展開プロセスが失敗する可能性がありますが、数回試行すると成功する可能性があります。次のログが表示されれば、起動は成功したことになります。

 2022-01-08T22:27:58,726+0000 [main] INFO org.apache.pulsar.broker.PulsarService - メッセージング サービスは準備完了、ブートストラップ サービス ポート = 8080、ブローカー URL = pulsar://localhost:6650、クラスター = スタンドアロン

ローカルのシングルノードクラスタが起動すると、public/defaultという名前空間が作成されます。

パルサークライアント

現在、Pulsar は以下を含む複数の言語でクライアントをサポートしています。

Java クライアント Go クライアント Python クライアント C++ クライアント Node.js クライアント WebSocket クライアント C# クライアント

SpringBoot 構成

SpringBoot を使用して Pulsar クライアントを統合します。まず、Pulsar クライアントの依存関係を導入します。コードは次のとおりです。

 <依存関係>
    <グループ ID>org.apache.pulsar</グループ ID>
    <artifactId>パルサークライアント</artifactId>
    <バージョン>2.9.1</バージョン>
 </依存関係>

次に、プロパティファイルに構成を追加します。

 # パルサーアドレス
pulsar.url=pulsar://192.168.59.155:6650
 ＃ トピック
pulsar.topic=テストトピック
# 消費者グループ 
 pulsar.subscription=トピックグループ

クライアントの作成

クライアントの作成は非常に簡単で、コードは次のとおりです。

クライアント = PulsarClient.builder()
                .serviceUrl(URL) は、
                。建てる（）;

上記の URL は、プロパティファイルで定義されている pulsar.url です。

クライアントを作成するときに、クラスターが正常に起動されなかったとしても、実際にはまだクラスターに接続されていないため、プログラムはエラーを報告しません。

プロデューサーを作成

プロデューサー = client.newProducer()
                .topic(トピック)
                .compressionType(圧縮タイプ.LZ4)
                .sendTimeout(0, 時間単位.SECONDS)
                .enableBatching( true )
                .batchingMaxPublishDelay(10, 時間単位.ミリ秒)
                .バッチ処理最大メッセージ数(1000)
                .maxPendingMessages(1000)
                .blockIfQueueFull( true )
                .roundRobinRouterBatchingPartitionSwitchFrequency(10)
                .batcherBuilder( BatcherBuilder.DEFAULT )
                。作成する（）;

プロデューサーを作成すると、実際にクラスターに接続します。クラスターに問題がある場合は、接続エラーが報告されます。

プロデューサーを作成するためのパラメータについて次に説明します。

トピック: プロデューサーが書きたいトピック。

compressionType: 圧縮戦略。現在、4 つの戦略がサポートされています (NONE、LZ4、ZLIB、ZSTD)。 Pulsar 2.3 以降では、この戦略はコンシューマーバージョンが 2.3 以上の場合にのみ有効になります。

sendTimeout: タイムアウト期間。プロデューサーがタイムアウト期間内に ACK を受信しない場合、メッセージを再送信します。

enableBatching: メッセージのバッチ処理を有効にするかどうか。デフォルト値は true です。このパラメータは、非同期送信 (sendAsync) の場合にのみ有効です。同期送信を選択した場合は無効になります。

batchingMaxPublishDelay: メッセージをバッチで送信する期間。ここでは10msと定義します。バッチ処理時間を設定すると、メッセージの数によって影響を受けなくなることに注意してください。バッチ送信では、送信するバッチメッセージを 1 つのネットワークパケットに入れて送信するため、ネットワーク IO 回数が削減され、ネットワークカードの送信効率が大幅に向上します。

batchingMaxMessages: バッチで送信されるメッセージの最大数。

maxPendingMessages: ブローカーからの ACK の受信を待機しているメッセージキューの最大長。キューがいっぱいの場合、blockIfQueueFull 値が true に設定されていない限り、プロデューサーのすべての sendAsync および send 呼び出しは失敗します。

blockIfQueueFull: プロデューサーがメッセージを送信すると、まずそのメッセージがローカルキューキャッシュに格納されます。キャッシュがいっぱいになると、メッセージの送信がブロックされます。

roundRobinRouterBatchingPartition-SwitchFrequency: メッセージを送信するときにキーが指定されていない場合、デフォルトでラウンドロビン方式を使用してメッセージが送信されます。ラウンドロビン方式を使用する場合、パーティション切り替え周期は (頻度 * batchingMaxPublishDelay) になります。

消費者の創造

Pulsar の消費モデルは次のとおりです。

図からわかるように、コンシューマーは消費する前にサブスクリプションをバインドする必要があります。

 Consumer = client.newConsumer()
        .topic(トピック)
        .subscriptionName(サブスクリプション)
        .subscriptionType(サブスクリプションタイプ.Shared)
        .subscriptionInitialPosition(サブスクリプションの初期位置.Earliest)
        .negativeAck再配信遅延(60, TimeUnit.SECONDS)
        .レシーバーキューサイズ(1000)
        。購読する（）;

以下では、コンシューマーを作成するためのパラメータについて説明します。

トピック: コンシューマーがサブスクライブするトピック。

subscriptionName: コンシューマーが関連付けるサブスクリプションの名前。

subscriptionType: サブスクリプションの種類。 Pulsar は次の 4 種類のサブスクリプションをサポートしています。

排他: 排他モード。同じトピックに対してコンシューマーは 1 つしか存在できません。コンシューマーが複数存在する場合、エラーが発生します。フェイルオーバー: 災害復旧モード。同じトピックには複数のコンシューマーが存在する可能性がありますが、消費できるのは 1 つのコンシューマーだけです。他のコンシューマーはフェイルオーバーバックアップとして機能します。現在のコンシューマーに障害が発生した場合、バックアップコンシューマーの 1 つが選択されて消費されます。以下のように表示されます。

共有: 共有モードでは、同じトピックを複数のコンシューマーがサブスクライブして使用できます。メッセージはラウンドロビンポーリングメカニズムを通じてさまざまなコンシューマーに配信され、各メッセージは 1 つのコンシューマーにのみ配信されます。コンシューマーが切断された場合、そのコンシューマーに送信されたメッセージが消費されていない場合、これらのメッセージは他の存続しているコンシューマーに再配布されます。以下のように表示されます。

Key_Shared: メッセージとコンシューマーの両方がキーにバインドされ、メッセージは同じキーにバインドされたコンシューマーにのみ送信されます。新しいコンシューマーが接続を確立するか、コンシューマーが切断された場合、一部のメッセージのキーを更新する必要があります。 Shared モードと比較すると、Key_Shared の利点は、同じキーのメッセージの順序を確保しながら、コンシューマーが同時にメッセージを消費できることです。以下のように表示されます。

subscriptionInitialPosition: 新しいサブスクリプションを作成するときに消費を開始する場所。次の 2 つのオプションがあります。

最新: 最新のメッセージから消費を開始します。最古: 最も古いメッセージから消費を開始します。

negativeAckRedeliveryDelay: 消費失敗後にブローカーが再送信する間隔。

受信キューサイズ: 受信メソッドが呼び出される前に蓄積できるメッセージの最大数。 0 に設定すると、ブローカーから一度にプルされるメッセージは 1 つだけになります。共有モードでは、バッチメッセージが複数のコンシューマーに送信され、他のコンシューマーがアイドル状態になるのを防ぐために、receiverQueueSize は 0 に設定されます。

コンシューマーがメッセージを受信する方法は、同期単一、同期バッチ、非同期単一、非同期バッチの 4 つがあります。コードは次のとおりです。

メッセージ message = consumer.receive()
 CompletableFuture<Message> メッセージ = consumer.receiveAsync();
メッセージ message = consumer.batchReceive();
 CompletableFuture<Messages> メッセージ = consumer.batchReceiveAsync();

バッチ受信の場合は、バッチ受信戦略も設定できます。コードは次のとおりです。

 Consumer = client.newConsumer()
    .topic(トピック)
    .subscriptionName(サブスクリプション)
        .batchReceivePolicy(BatchReceivePolicy.builder()
        .最大メッセージ数(100)
        .maxNumBytes(1024 * 1024)
        .timeout(200, 時間単位.ミリ秒)
        。建てる（））
    。購読する（）;

コード内のパラメータは次のように記述されます。

maxNumMessages: バッチで受信されるメッセージの最大数。 maxNumBytes: 受信したバッチメッセージのサイズ。ここでは 1MB です。

テスト

まず、Producer がメッセージを送信するためのコードを次のように記述します。

パブリックvoid sendMsg(文字列キー、文字列データ) {
    CompletableFuture<MessageId> future = producer.newMessage()
        。キー(キー)
        .value(data.getBytes()).sendAsync();
    future.handle((v, ex) -> {
        if (ex == null ) {
            logger.info( "メッセージは正常に送信されました、キー:{}、メッセージ:{}" 、キー、データ);
        }それ以外{
            logger.error( "メッセージの送信に失敗しました。キー:{}、メッセージ:{}" 、キー、データ);
        }
戻る ヌル;
    });
    将来参加();
    logger.info( "メッセージの送信が完了しました、キー:{}、メッセージ:{}" 、キー、データ);
 }

次に、メッセージを消費するための Consumer コードを次のように記述します。

パブリックvoid start() は例外をスローします{
    （真）の間{
        メッセージ message = consumer.receive();
        文字列キー= message.getKey();
        文字列データ = new String(message.getData());
        文字列トピック = message.getTopicName();
        StringUtils.isNotEmpty(データ)の場合{
            試す{
                logger.info( "受信メッセージ、トピック:{}、キー:{}、データ:{}" 、トピック、キー、データ);
            }catch(例外 e){
                logger.error( "メッセージ受信例外、トピック:{}、キー:{}、データ:{}" 、トピック、キー、データ、e);
            }
        }
        消費者はメッセージを確認します。
    }
 }

最後に、Producer を呼び出してメッセージを送信する Controller クラスを記述します。コードは次のとおりです。

 @RequestMapping( "/送信" )
 @レスポンス本文
パブリック文字列送信(@RequestParam 文字列キー、@RequestParam 文字列データ) {
    logger.info( "メッセージ送信要求を受信しました、キー:{}、値:{}" 、キー、データ);
    pulsarProducer.sendMsg(キー、データ );
戻る  "成功" ;
 }

プロデューサーを呼び出して、キー = key1、データ = data1 のメッセージを送信します。具体的な操作は、ブラウザに次の URL を入力して Enter キーを押すことです。

 http://192.168.157.1:8083/pulsar/send?キー=key1&データ =data1

コンソールに次のログ出力が表示されます。

 2022-01-08 22:42:33,199 [pulsar-client-io-6-1] [INFO] boot.pulsar.PulsarProducer - メッセージを正常に送信しました。キー: key1、メッセージ: data1
 2022-01-08 22:42:33,200 [http-nio-8083- exec -1] [INFO] boot.pulsar.PulsarProducer - メッセージの送信が完了しました。キー: key1、メッセージ: data1
 2022-01-08 22:42:33,232 [スレッド-22] [INFO] boot.pulsar.PulsarConsumer - メッセージを受信しました。トピック:persistent:// public / default /testTopic、キー: key1、データ: data1
 2022-01-08 22:43:14,498 [pulsar-timer-5-1] [INFO] org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - [testTopic] [topicGroup] [7def6] プリフェッチされたメッセージ: 0 --- 受信した消費スループット: 0.02 メッセージ/秒 --- 0.00 メガビット/秒 --- 確認応答送信レート: 0.02 確認応答/秒 --- 失敗したメッセージ: 0 --- バッチメッセージ: 0 --- 失敗した確認応答: 0  
 2022-01-08 22:43:14,961 [pulsar-timer-9-1] [INFO] org.apache.pulsar.client.impl.ProducerStatsRecorderImpl - [testTopic] [standalone-9-0] 保留中のメッセージ: 0 --- パブリッシュ スループット: 0.02 メッセージ/秒 --- 0.00 メガビット/秒 --- レイテンシ: 中: 69.000 ミリ秒 - 95 パーセント: 69.000 ミリ秒 - 99 パーセント: 69.000 ミリ秒 - 99.9 パーセント: 69.000 ミリ秒 - 最大: 69.000 ミリ秒 --- 確認応答受信率: 0.02 確認応答/秒 --- 失敗したメッセージ: 0