自研 Pulsar Starter：Winfun-Pulsar-Spring-Boot-Starter

[[420613]]

マイルストーン

1. 背景

新世代のクラウドネイティブ メッセージ キューとして、Pulsar は開発者の間でますます人気が高まっています。現在のプロジェクトのほとんどは SpringBoot をベースに開発されていますが、今のところ Pulsar 用の人気があり成熟したスターターは存在しないため、Pulsar API を一般的に使用する際に大量の重複コードを回避するために、独自にスターターを作成する必要があります。

2. デザインアイデア

これは設計の最初のバージョンなので、シンプルさから始め、最初から複雑になりすぎないようにし、Pulsar API のネイティブ機能を維持するようにします。

2.1、パルサークライアント

プロデューサーとコンシューマーの両方が PulsarClient によって作成されることは誰もが知っています。

もちろん、PulsarClient はビジネス ニーズに応じて多くのパラメータをカスタマイズできますが、設計の最初のバージョンでは、より一般的に使用されるパラメータのみがサポートされます。

当社のコンポーネントは次の機能をサポートしています。

• PulsarClient パラメータ設定の外部化をサポートします。パラメータはapplicatin.propertiesで設定できます。
• 構成プロンプト情報を提供するために、applicatin.properties をサポートします。
• 外部構成ファイルを読み取り、パラメータに従って PulsarClient をインスタンス化し、IOC コンテナに挿入します。

2.2 プロデューサー

プロデューサーはメッセージを送信するコンポーネントです。

• ここでは、要件に応じて対応する Producer インスタンスを作成できるテンプレート クラスを提供します。
• Producer インスタンスの繰り返し作成を回避するために、Topic<->Producer 関係のキャッシュをサポートします。
• 同期/非同期メッセージ送信をサポートします。

2.3 消費者

コンシューマーはメッセージを消費するコンポーネントです。

• ここでは抽象クラスを提供します。開発者は、この実装クラスを統合し、メッセージを消費するための論理メソッドである doReceive メソッドを実装するだけで済みます。
• さらに、トピック、テナント、名前空間などのカスタム Consmuer 構成をサポートするカスタム アノテーションも提供されます。
• 上記のカスタム アノテーションを実装クラスに追加すると、コンポーネントは対応する Consumer インスタンスを自動的に認識して生成します。
• 同期/スレッド プールの非同期消費をサポートします。

3. 使用例

3.1.依存関係の導入

<依存関係>  
 <グループID> io.github.howinfun</グループID>  
 <artifactId> winfun- pulsar - spring -boot-スターター</artifactId>  
 <バージョン> 1.1.0</バージョン>  
 </依存関係>  

3.2.設定を追加

pulsar.service-url=pulsar://127.0.0.1:6650
 pulsar.tenant=winfun
 pulsar.namespace=研究
パルサー操作タイムアウト=30
 pulsar.io-スレッド=10
 pulsar.リスナースレッド=10

3.3 メッセージの送信

/**
 * メッセージを送信
 * @著者: winfun
 **/
 @レストコントローラ
@RequestMapping( "メッセージ" )
パブリッククラスMessageController{ 
 
    オートワイヤード
    プライベート PulsarTemplate pulsarTemplate;
    オートワイヤード
    プライベート PulsarProperties pulsarProperties; 
 
    /***
     * 指定されたトピックにメッセージを送信する
     * @author ウィンファン
     * @param トピック トピック
     * @param メッセージ メッセージ
     * @return {@link 文字列 }
     **/
    @GetMapping( "/{トピック}/{メッセージ}" )
パブリック文字列 send(@PathVariable( "topic" ) 文字列 topic,@PathVariable( "msg" ) 文字列 msg) 例外をスローします {
        this.pulsarTemplate.createBuilder().persistent(ブール値.TRUE )
                .tenant(this.pulsarProperties.getTenant())
                .namespace(this.pulsarProperties.getNamespace())
                .topic(トピック)
                .send(メッセージ);
戻る  "成功" ;
    }
 }

3.4.消費メッセージ

/**
 * @著者: winfun
 * @日付: 2021/8/20 午後 8:13
 **/
翻訳者
@PulsarListener(トピック = { "テストトピック2" },
                スレッドプール = @ThreadPool(
                                        コアスレッド = 2、
                                        最大コアスレッド数 = 3、
                                        threadPoolName = "テストスレッドプール" ))
パブリッククラス ConsumerListener は BaseMessageListener を拡張します { 
 
    /**
     * 消費メッセージ
     * @param 消費者 消費者
     * @param msg メッセージ
     */
    @オーバーライド
    保護されたvoid doReceived(Consumer<String> consumer, Message<String> msg) {
        log.info( "消費成功メッセージ: {}" , msg.getValue());
        試す {
            消費者はメッセージを確認します。
        } キャッチ (PulsarClientException e) {
            e.printStackTrace();
        }
    } 
 
    /***
     * 非同期消費を有効にするかどうか
     * @return {@link ブール値 }
     **/
    @オーバーライド
パブリックブールenableAsync() {
 Boolean.TRUEを返します。
    }
 }

4. ソースコード

ここではソースコードを分析しませんが、Github で確認できます。コードに関してご提案やご意見がございましたら、ぜひ MR をご提出ください。