🥟 👨🏾 🍷 Spring統合-動的データフロー 🚎 🔞 👈🏽

花火、ハーブル！今日はかなり特定の領域、Spring Integrationフレームワークを使用したデータストリーミングと、アプリケーションコンテキストで事前に初期化せずに実行時にこれらのフローを作成する方法を分析します。完全なサンプルアプリケーションはGitaにあります。

前書き

Spring Integrationは、異なるプロトコルのアダプター/メッセージチャネル（条件付きキュー）に基づく統合システム間のアダプターの下でメッセージングメカニズムを使用するエンタープライズ統合フレームワーク（EIP）です。有名な類似体は、キャメル、ミュール、ニフィです。

テストケースから、受信したリクエストパラメータを読み取ることができるRESTサービスを作成し、たとえばpostgresなどのデータベースに移動し、ソースから受信したパラメータに従ってテーブルデータを更新してフェッチし、結果をキューに送信します（リクエスト/応答）、そして異なるリクエストパスで複数のインスタンスを作成します。

通常、データフロー図は次のようになります。

次に、REST制御コンポーネント/スレッドエンドポイントを使用して、IntegrationFlowContextを使用して、タンバリンであまりダンスせずにこれを簡単に行う方法を示します。すべてのメインプロジェクトコードはリポジトリに配置されます。ここでは、いくつかのクリッピングのみを示します。ねえ、猫の下で気になる人はどうぞ。

ツール

デフォルトの依存関係ブロックから始めましょう。基本的に、フローとコンポーネントの管理のRESTイデオロギー、スプリング統合のために、チャネルとアダプタに基づいてケースを作成するために、スプリングブートプロジェクトが必要です。

そして、私たちはすぐにケースを再現するために他に何が必要か考えます。コアの依存関係に加えて、サブプロジェクト-Integration-http、integration-jdbc、integration-groovy（Goovyスクリプトに基づいて動的にカスタマイズ可能なデータコンバーターを提供）が必要です。それとは別に、この例では不要なgroovyコンバーターは使用しないと言いますが、外部からカスタマイズする機能を提供します。

依存関係リスト

 <!-- Spring block -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-commons</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-groovy</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-http</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-jdbc</artifactId>
        </dependency>

        <!-- Db block -->
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
        </dependency>

        <dependency>
            <groupId>com.zaxxer</groupId>
            <artifactId>HikariCP</artifactId>
        </dependency>

        <!-- Utility block -->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>

        <dependency>
            <groupId>org.reflections</groupId>
            <artifactId>reflections</artifactId>
            <version>0.9.12</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.4</version>
            <scope>provided</scope>
        </dependency>

内部キッチン

必要なシステムコンポーネント（ラッパー/モデル）の作成に移りましょう。チャネル、Bean、httpInboundGateway、ハンドラー、jdbcOutboundGateway、および結果モデルが必要です。

Bean-アダプタ、スレッドに必要なヘルパーオブジェクト
channel-ストリームコンポーネントとの間でメッセージを配信するためのチャネル
httpInboundGateway-さらに処理するためのデータを含むリクエストをさらに送信するhttpアクセスポイント
ハンドラー-ハンドラーの一般的なタイプ（グルーブトランスフォーマー、さまざまなアダプターなど）
jdbcOutboundGateway-JDBCアダプター
result-特定のチャネルに情報を送信するためのハンドラー

パラメータを格納し、ストリーム全体のコンポーネントを正しく初期化するラッパーが必要なので、コンポーネントストアをすぐに作成して追加します。 JSONコンバーターの機能->定義モデル。私の場合、ジャクソンとオブジェクトを使用したフィールドの直接マッピングは適用できませんでした-特定の通信プロトコル用にもう1台自転車があります。アノテーションを使用して

、すぐにうまくやりましょう：StreamComponent-クラスをストリームコンポーネントのチューニングモデルとして識別する責任があり、サービス情報（コンポーネントの名前、コンポーネントのタイプ、コンポーネントがネストされているかどうか、説明）があります。 SettingClass-スーパークラスフィールドのスキャンや値の初期化時のフィールドの無視など、モデルをスキャンするための追加オプションを担当します。

SettingValue-クラスフィールドを外部からカスタマイズ可能なものとして識別し、JSONの名前付け設定、説明、タイプコンバーター、情報フィールドの必須フィールドフラグと内部オブジェクトフラグで識別します。RESTコントローラーのモデルを操作するための

コンポーネントストレージマネージャー

ヘルパーメソッド

基本モデル-一連の補助フィールド/モデルメソッドによる抽象化

現在のフロー構成モデル

マッパーJSON->定義モデル

作業のメイングラウンドが準備されました。次に、ストリームのライフサイクル、ストレージ、および初期化を担当するサービスの実装に直接取り掛かります。同じ名前の1つのストリームを複数のインスタンスに並列化できるという考えをすぐに説明します。つまり、フローのすべてのコンポーネントに一意の識別子（GUID）を作成する必要があります。そうしないと、アプリケーションコンテキストで他のシングルトンコンポーネント（Bean、チャネルなど）との衝突が発生する可能性があります。ただし、最初に2つのコンポーネント（httpとjdbc）のマッパーを作成しましょう。ストリーム自体のコンポーネント（HttpRequestHandlerEndpointSpecおよびJdbcOutboundGateway）に対して以前に作成されたモデルの増分。

HttpRegistry

JdbcRegistry

中央管理サービス（StreamDeployingService）ワーカー/非アクティブを保存する機能を実行し、新しいワーカーを登録し、アプリケーションコンテキストからスレッドを完全に開始、停止、削除します。サービスの重要な機能はIntegrationFlowBuilderRegistry依存関係の実装です。これは、アプリケーションのダイナミクスを作成するのに役立ちます（おそらく、これらの構成xmlファイルまたはキロメートルのDSLクラスを覚えておいてください）。ストリームの仕様によると、これは常にインバウンドコンポーネントまたはチャネルから開始する必要があるため、registerStreamContextメソッドの実装でこれを考慮します。

そして補助マネージャー（IntegrationFlowBuilderRegistry）。これは、フローコンポーネントへのモデルのマッパーとIntegrationFlowBuilderを使用したフロー自体の初期化の両方の機能を実行します。また、フローパイプラインにログハンドラー、フローチャネルメトリックを収集するためのサービス（切り替え可能なオプション）、およびGroovy実装に基づくフローメッセージコンバーターの可能な実装を実装しました（突然この例が販売の基礎になった場合、フローの初期化の段階でgroovyスクリプトのプリコンパイルを行う必要があります）、RAMで負荷テストを実行すると、コアの数と電力に関係なく）。モデルのログステージとログレベルのパラメーターの構成に応じて、コンポーネントからコンポーネントへのメッセージの各送信後にアクティブになります。モニタリングはapplication.ymlのパラメーターによって有効化および無効化されます：

monitoring:
  injectction:
    default: true

これで、動的データ処理フローを初期化するためのすべてのメカニズムが揃ったので、RabbitMQ、Kafka、Tcp、Ftpなどのさまざまなプロトコルおよびアダプター用のマッパーを追加で作成できます。さらに、ほとんどの場合、自分の手で何かを書く必要はありません（もちろん、構成モデルと補助メソッドを除く）-かなり多数のコンポーネントがリポジトリにすでに存在しています。

最終段階は、既存のシステムコンポーネントに関する情報の取得、フローの管理、およびメトリックの取得のためのコントローラーの実装です。

ComponentsController-人間が読めるモデルのすべてのコンポーネントに関する情報と、名前とタイプごとに1つのコンポーネントを提供します。

StreamController -完全なフロー管理、つまり新しいJSONモデルの初期化、識別子によるメトリックの開始、停止、削除、発行を提供します。

最終製品

結果のアプリケーションを生成し、テストケースをJSON形式で記述します。

サンプルデータストリーム

データベース初期化スクリプト：

CREATE TABLE IF NOT EXISTS account_data
(
    id          INT                      NOT NULL,
    accountname VARCHAR(45)              NOT NULL,
    password    VARCHAR(128),
    email       VARCHAR(255),
    last_ip     VARCHAR(15) DEFAULT NULL NOT NULL
);

CREATE UNIQUE INDEX account_data_username_uindex
    ON account_data (accountname);

ALTER TABLE account_data
    ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY (
        SEQUENCE NAME account_data_id_seq
            START WITH 1
            INCREMENT BY 1
            NO MINVALUE
            NO MAXVALUE
            CACHE 1
        );

ALTER TABLE account_data
    ADD CONSTRAINT account_data_pk
        PRIMARY KEY (id);

CREATE TABLE IF NOT EXISTS account_info
(
    id             INT NOT NULL,
    banned         BOOLEAN  DEFAULT FALSE,
    premium_points INT      DEFAULT 0,
    premium_type   SMALLINT DEFAULT -1
);

ALTER TABLE account_info
    ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY (
        SEQUENCE NAME account_info_id_seq
            START WITH 1
            INCREMENT BY 1
            NO MINVALUE
            NO MAXVALUE
            CACHE 1
        );

ALTER TABLE account_info
    ADD CONSTRAINT account_info_account_data_id_fk FOREIGN KEY (id) REFERENCES account_data (id)
        ON UPDATE CASCADE ON DELETE CASCADE;

ALTER TABLE account_info
    ADD CONSTRAINT account_info_pk
        PRIMARY KEY (id);



INSERT INTO account_data (accountname, password, email, last_ip)
VALUES ('test', 'test', 'test@test', '127.0.0.1');
INSERT INTO account_info (banned, premium_points, premium_type)
VALUES (false, 1000, 1);

: order — , .. , . ( ). — .

{
  "flowName": "Rest Postgres stream",
  "components": [
    {
      "componentName": "bean",
      "componentType": "other",
      "componentParameters": {
        "id": "pgDataSource",
        "bean-type": "com.zaxxer.hikari.HikariDataSource",
        "property-args": [
          {
            "property-name": "username",
            "property-value": "postgres"
          },
          {
            "property-name": "password",
            "property-value": "postgres"
          },
          {
            "property-name": "jdbcUrl",
            "property-value": "jdbc:postgresql://localhost:5432/test"
          },
          {
            "property-name": "driverClassName",
            "property-value": "org.postgresql.Driver"
          }
        ]
      }
    },
    {
      "componentName": "message-channel",
      "componentType": "source",
      "componentParameters": {
        "id": "jdbcReqChannel",
        "order": 1,
        "channel-type": "direct",
        "max-subscribers": 1000
      }
    },
    {
      "componentName": "message-channel",
      "componentType": "source",
      "componentParameters": {
        "id": "jdbcRepChannel",
        "order": 1,
        "channel-type": "direct"
      }
    },
    {
      "componentName": "http-inbound-gateway",
      "componentType": "source",
      "componentParameters": {
        "order": 2,
        "http-inbound-supported-methods": [
          "POST"
        ],
        "payload-type": "org.genfork.integration.model.request.http.SimpleJdbcPayload",
        "log-stages": true,
        "log-level": "INFO",
        "request-channel": "jdbcReqChannel",
        "reply-channel": "jdbcRepChannel"
      }
    },
    {
      "componentName": "handler",
      "componentType": "processor",
      "componentParameters": {
        "order": 3,
        "handler-definition": {
          "componentName": "jdbc-outbound-adapter",
          "componentType": "app",
          "componentParameters": {
            "data-source": "pgDataSource",
            "query": "SELECT accountname, password, email, last_ip, banned, premium_points, premium_type FROM account_data d INNER JOIN account_info i ON d.id = i.id WHERE d.id = :payload.accountId",
            "update-query": "UPDATE account_info SET banned = true WHERE id = :payload.accountId",
            "jdbc-reply-channel": "jdbcRepChannel",
            "log-stages": true,
            "log-level": "INFO"
          }
        }
      }
    },
    {
      "componentName": "result",
      "componentType": "app",
      "componentParameters": {
        "order": 4,
        "cancel": false,
        "result-channel": "jdbcRepChannel"
      }
    }
  ]
}

テスト：

1）

POST / stream / deploy メソッドを使用して新しいストリームを初期化します。ここで、JSONはリクエストの本文に含まれます。

応答として、すべてが正しい場合、システムは送信する必要があります。そうでない場合、エラーメッセージが表示されます。

{
    "status": "SUCCESS", -  
    "streamId": "2bf65d9d-97c6-4199-86aa-0c808c25071b" -  
}

2）次のメソッドを使用して開始を開始し

ます：GET / stream / 2bf65d9d-97c6-4199-86aa-0c808c25071b / start。ここで、初期化されたストリームの識別子を先に示します。

応答として、すべてが正しい場合、システムは送信する必要があります。そうでない場合、エラーメッセージが表示されます。

{
    "status": "SUCCESS", -  
}

3）システム内の識別子でストリームを呼び出していますか？どのように、何を、どこで-HttpRegistryモデルのマッパーで、条件を記述しました

Http.inboundGateway(localPath != null ? localPath : String.format("/stream/%s/call", uuid))

ここで、http-inbound-pathパラメーターが考慮され、コンポーネントの構成で明示的に指定されていない場合は無視され、システムコールパスが設定されます。私たちの場合、これは次のようになります：

POST / stream / ece4d4ac-3b46-4952-b0a6-8cf334074b99 / call-ストリームの識別子が存在し、リクエストの本文がここにあります：

{
    "accountId": 1
}

それに応じて、リクエストの処理段階が正しく機能していれば、account_dataテーブルとaccount_infoテーブルのレコードのフラット構造を受け取ります。

{
    "accountname": "test",
    "password": "test",
    "email": "test@test",
    "last_ip": "127.0.0.1",
    "banned": true,
    "premium_points": 1000,
    "premium_type": 1
}

JdbcOutboundGatewayアダプターの特性は、update-queryパラメーターを指定した場合、追加のハンドラーが登録され、最初にデータを更新してから、queryパラメーターによってのみフェッチするというものです。

同じパスを手動で指定すると、いくつかのインスタンスでストリームへのアクセスポイントとしてHttpInboundGatewayを使用してコンポーネントを起動する可能性がなくなります。これは、システムが同様のパスを登録できないためです。

4）GETメソッド/ストリーム/ 2bf65d9d-97c6-4199-86aa-0c808c25071b /メトリックを使用してメトリックを見てみましょう

応答内容

, / , / / :

[
    {
        "streamId": "2bf65d9d-97c6-4199-86aa-0c808c25071b",
        "channelName": "application.Rest Postgres stream_2bf65d9d-97c6-4199-86aa-0c808c25071b_jdbcReqChannel",
        "sendDuration": {
            "count": 1,
            "min": 153.414,
            "max": 153.414,
            "mean": 153.414,
            "standardDeviation": 0.0,
            "countLong": 1
        },
        "maxSendDuration": 153.414,
        "minSendDuration": 153.414,
        "meanSendDuration": 153.414,
        "meanSendRate": 0.001195117818082359,
        "sendCount": 1,
        "sendErrorCount": 0,
        "errorRate": {
            "count": 0,
            "min": 0.0,
            "max": 0.0,
            "mean": 0.0,
            "standardDeviation": 0.0,
            "countLong": 0
        },
        "meanErrorRate": 0.0,
        "meanErrorRatio": 1.1102230246251565E-16
    },
    {
        "streamId": "2bf65d9d-97c6-4199-86aa-0c808c25071b",
        "channelName": "application.2bf65d9d-97c6-4199-86aa-0c808c25071b.channel#2",
        "sendDuration": {
            "count": 1,
            "min": 0.1431,
            "max": 0.1431,
            "mean": 0.1431,
            "standardDeviation": 0.0,
            "countLong": 1
        },
        "maxSendDuration": 0.1431,
        "minSendDuration": 0.1431,
        "meanSendDuration": 0.1431,
        "meanSendRate": 0.005382436008121413,
        "sendCount": 1,
        "sendErrorCount": 0,
        "errorRate": {
            "count": 0,
            "min": 0.0,
            "max": 0.0,
            "mean": 0.0,
            "standardDeviation": 0.0,
            "countLong": 0
        },
        "meanErrorRate": 0.0,
        "meanErrorRatio": 0.0
    },
    {
        "streamId": "2bf65d9d-97c6-4199-86aa-0c808c25071b",
        "channelName": "application.Rest Postgres stream_2bf65d9d-97c6-4199-86aa-0c808c25071b_jdbcRepChannel",
        "sendDuration": {
            "count": 1,
            "min": 0.0668,
            "max": 0.0668,
            "mean": 0.0668,
            "standardDeviation": 0.0,
            "countLong": 1
        },
        "maxSendDuration": 0.0668,
        "minSendDuration": 0.0668,
        "meanSendDuration": 0.0668,
        "meanSendRate": 0.001195118373693797,
        "sendCount": 1,
        "sendErrorCount": 0,
        "errorRate": {
            "count": 0,
            "min": 0.0,
            "max": 0.0,
            "mean": 0.0,
            "standardDeviation": 0.0,
            "countLong": 0
        },
        "meanErrorRate": 0.0,
        "meanErrorRatio": 1.1102230246251565E-16
    }
]

結論

したがって、他のシステムとの統合のためにアプリケーションで毎回追加の手動ハンドラー（パイプライン）を作成するよりも、さまざまなシステムと統合するためのアプリケーションを作成するのに少し時間と労力を費やして、コードをそれぞれ200〜500行作成する方法が示されました。

現在の例では、スレッドの依存関係（ビン、チャネルなど）間のアプリケーションのグローバルコンテキストでの衝突を回避する一意の識別子を使用して、複数のインスタンスの同じタイプのスレッドの作業を並列化できます。

さらに、プロジェクトを開発できます。

ストリームをデータベースに保存します。
SpringおよびSpring統合コミュニティが提供するすべての統合コンポーネントをサポートします。
スケジュールに従ってスレッドを使用して作業するワーカーを作成します。
条件付きの「マウスとコンポーネントキューブ」でストリームを構成するための適切なUIを作成します（ちなみに、この例はgithub.com/spring-cloud/spring-cloud-dataflow-uiプロジェクト用に部分的にシャープにされました）。

そしてもう一度、リポジトリへのリンクを複製します。

Spring統合-動的データフロー

前書き

ツール

内部キッチン

最終製品

結論

More articles: