JDBC サポート
Spring Integration は、データベースクエリを使用してメッセージを送受信するためのチャネルアダプターを提供します。これらのアダプターを通じて、Spring Integration はプレーンな JDBC SQL クエリだけでなく、ストアドプロシージャとストアドファンクションコールもサポートします。
この依存関係をプロジェクトに含める必要があります。
デフォルトでは、次の JDBC コンポーネントが使用可能です。
Spring Integration JDBC モジュールは JDBC メッセージストアも提供します。
受信チャネルアダプター
受信チャネルアダプターの主な機能は、SQL SELECT クエリを実行し、結果セットをメッセージに変換することです。メッセージペイロードは結果セット全体（List として表される）であり、リスト内のアイテムの型は行マッピング戦略に依存します。デフォルトの戦略は、クエリ結果の各行に対して Map を返す汎用マッパーです。オプションで、RowMapper インスタンスへの参照を追加することでこれを変更できます（行マッピングの詳細については、Spring JDBC の資料を参照してください）。
SELECT クエリ結果の行を個々のメッセージに変換する場合は、ダウンストリームスプリッターを使用できます。 |
受信アダプターには、JdbcTemplate インスタンスまたは DataSource への参照も必要です。
メッセージを生成する SELECT ステートメントの他に、アダプターには、次のポーリングで表示されないようにレコードを処理済みとしてマークする UPDATE ステートメントもあります。更新は、元の選択からの ID のリストによってパラメーター化できます。デフォルトでは、これは命名規則によって行われます（id と呼ばれる入力結果セットの列は、id と呼ばれる更新のパラメーターマップのリストに変換されます）。次の例では、更新クエリと DataSource 参照を使用して受信チャネルアダプターを定義しています。
<int-jdbc:inbound-channel-adapter query="select * from item where status=2"
channel="target" data-source="dataSource"
update="update item set status=10 where id in (:id)" /> 更新クエリのパラメーターは、パラメーター名の前にコロン（:）を付けて指定します（前述の例では、ポーリングされた結果セットの各行に適用される式です）。これは、Spring JDBC の名前付きパラメーター JDBC サポートの標準機能であり、Spring Integration で採用された規約（ポーリングされた結果リストへの射影）と組み合わせたものです。基本的な Spring JDBC の機能は、利用可能な表現を制限しますが（たとえば、ピリオド以外のほとんどの特殊文字は許可されません）、ターゲットは通常、Bean のパスでアドレス指定可能なオブジェクトのリスト（1 つのリストの場合もあります）であるため、これは過度に制限されるものではありません。 |
パラメーター生成戦略を変更するには、SqlParameterSourceFactory をアダプターに挿入して、デフォルトの動作をオーバーライドできます（アダプターには sql-parameter-source-factory 属性があります）。Spring Integration は ExpressionEvaluatingSqlParameterSourceFactory を提供します。ExpressionEvaluatingSqlParameterSourceFactory は、クエリの結果を #root オブジェクトとして SpEL ベースのパラメーターソースを作成します。（update-per-row が true の場合、ルートオブジェクトは行です）。更新クエリに同じパラメーター名が複数回出現する場合、1 回だけ評価され、その結果がキャッシュされます。
選択クエリにパラメーターソースを使用することもできます。この場合、評価の対象となる「結果」オブジェクトがないため、（パラメーターソースファクトリを使用するのではなく）毎回単一のパラメーターソースが使用されます。バージョン 4.0 以降では、Spring を使用して、SpEL ベースのパラメーターソースを作成できます。以下に例を示します。
<int-jdbc:inbound-channel-adapter query="select * from item where status=:status"
channel="target" data-source="dataSource"
select-sql-parameter-source="parameterSource" />
<bean id="parameterSource" factory-bean="parameterSourceFactory"
factory-method="createParameterSourceNoCache">
<constructor-arg value="" />
</bean>
<bean id="parameterSourceFactory"
class="o.s.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
<property name="parameterExpressions">
<map>
<entry key="status" value="@statusBean.which()" />
</map>
</property>
</bean>
<bean id="statusBean" class="foo.StatusDetermination" /> 各パラメーター式の value は、任意の有効な SpEL 式にすることができます。式評価の #root オブジェクトは、parameterSource Bean で定義されたコンストラクター引数です。すべての評価に対して静的です（前の例では、空の String）。
バージョン 5.0 から、ExpressionEvaluatingSqlParameterSourceFactory に sqlParameterTypes を提供して、特定のパラメーターのターゲット SQL 型を指定できます。
次の例は、クエリで使用されているパラメーターの SQL 型を示しています。
<int-jdbc:inbound-channel-adapter query="select * from item where status=:status"
channel="target" data-source="dataSource"
select-sql-parameter-source="parameterSource" />
<bean id="parameterSource" factory-bean="parameterSourceFactory"
factory-method="createParameterSourceNoCache">
<constructor-arg value="" />
</bean>
<bean id="parameterSourceFactory"
class="o.s.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
<property name="sqlParameterTypes">
<map>
<entry key="status" value="#{ T(java.sql.Types).BINARY}" />
</map>
</property>
</bean>createParameterSourceNoCache ファクトリメソッドを使用します。それ以外の場合、パラメーターソースは評価の結果をキャッシュします。また、キャッシュが無効になっているため、選択クエリに同じパラメーター名が複数回表示される場合、出現するたびに再評価されることに注意してください。 |
ポーリングとトランザクション
受信アダプターは、通常の Spring Integration ポーラーを子要素として受け入れます。その結果、（他の用途の中でも）ポーリングの頻度を制御できます。JDBC を使用するためのポーラーの重要な機能は、次の例に示すように、トランザクションでポーリング操作をラップするオプションです。
<int-jdbc:inbound-channel-adapter query="..."
channel="target" data-source="dataSource" update="...">
<int:poller fixed-rate="1000">
<int:transactional/>
</int:poller>
</int-jdbc:inbound-channel-adapter>| ポーラーを明示的に指定しない場合、デフォルト値が使用されます。Spring Integration で通常のように、トップレベル Bean として定義できます。 |
上記の例では、データベースは 1000 ミリ秒ごと（または 1 秒間に 1 回）ポーリングされ、更新クエリと選択クエリは両方とも同じトランザクションで実行されます。トランザクションマネージャーの構成は表示されません。ただし、データソースを認識している限り、ポーリングはトランザクションです。一般的な使用例では、ダウンストリームチャネルが直接チャネル（デフォルト）であるため、エンドポイントは同じスレッドで呼び出されるため、同じトランザクションで呼び出されます。そのようにして、それらのいずれかが失敗した場合、トランザクションはロールバックし、入力データは元の状態に戻ります。
max-rows 対 max-messages-per-poll
JDBC 受信チャネルアダプターは、max-rows という属性を定義します。アダプターのポーラーを指定するときに、max-messages-per-poll というプロパティを定義することもできます。これら 2 つの属性は似ていますが、その意味はまったく異なります。
max-messages-per-poll はポーリング間隔ごとにクエリが実行される回数を指定しますが、max-rows は各実行で返される行の数を指定します。
通常の状況では、JDBC 受信チャネルアダプターを使用する際に、ポーラーの max-messages-per-poll プロパティを設定したくないでしょう。デフォルト値は 1 です。これは、JDBC 受信チャネルアダプターの receive() (Javadoc) メソッドがポーリング間隔ごとに 1 回だけ実行されることを意味します。
max-messages-per-poll 属性をより大きな値に設定すると、クエリが何度も連続して実行されます。max-messages-per-poll 属性の詳細については、受信チャネルアダプターの構成を参照してください。
対照的に、max-rows 属性は、0 より大きい場合、receive() メソッドによって作成されたクエリ結果セットから使用される最大行数を指定します。属性が 0 に設定されている場合、結果のメッセージにはすべての行が含まれます。属性のデフォルトは 0 です。
MySQL LIMIT または SQL Server TOP または Oracle の ROWNUM など、ベンダー固有のクエリオプションを介して結果セット制限を使用することをお勧めします。詳細については、特定のベンダーのドキュメントを参照してください。 |
送信チャネルアダプター
送信チャネルアダプターは受信の逆です。そのロールは、メッセージを処理し、それを使用して SQL クエリを実行することです。デフォルトでは、次の例に示すように、メッセージペイロードとヘッダーはクエリへの入力パラメーターとして使用できます。
<int-jdbc:outbound-channel-adapter
query="insert into foos (id, status, name) values (:headers[id], 0, :payload[something])"
data-source="dataSource"
channel="input"/> 上記の例では、input というラベルが付いたチャネルに到着するメッセージは、something のキーを持つマップのペイロードを持っているため、[] オペレーターはマップからその値を逆参照します。ヘッダーもマップとしてアクセスされます。
上記のクエリのパラメーターは、受信メッセージの Bean プロパティ式です（SpEL 式ではありません）。この動作は SqlParameterSource の一部であり、これは送信アダプターによって作成されるデフォルトのソースです。異なる動作を得るために、異なる SqlParameterSourceFactory を注入できます。 |
発信アダプターには、DataSource または JdbcTemplate への参照が必要です。SqlParameterSourceFactory を挿入して、各受信メッセージのクエリへのバインドを制御することもできます。
入力チャネルが直接チャネルの場合、送信アダプターはクエリを同じスレッドで実行するため、メッセージの送信者と同じトランザクション（存在する場合）を実行します。
SpEL 式を使用してパラメーターを渡す
ほとんどの JDBC チャネルアダプターの一般的な要件は、SQL クエリまたはストアドプロシージャまたは関数の一部としてパラメーターを渡すことです。前述のように、これらのパラメーターはデフォルトでは Bean プロパティ式であり、SpEL 式ではありません。ただし、SpEL 式をパラメーターとして渡す必要がある場合は、明示的に SqlParameterSourceFactory を注入する必要があります。
次の例では、ExpressionEvaluatingSqlParameterSourceFactory を使用してその要件を実現しています。
<jdbc:outbound-channel-adapter data-source="dataSource" channel="input"
query="insert into MESSAGES (MESSAGE_ID,PAYLOAD,CREATED_DATE) \
values (:id, :payload, :createdDate)"
sql-parameter-source-factory="spelSource"/>
<bean id="spelSource"
class="o.s.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
<property name="parameterExpressions">
<map>
<entry key="id" value="headers['id'].toString()"/>
<entry key="createdDate" value="new java.util.Date()"/>
<entry key="payload" value="payload"/>
</map>
</property>
</bean>詳細については、パラメーターソースの定義を参照してください。
PreparedStatement コールバックの使用
場合によっては、SqlParameterSourceFactory の柔軟性と疎結合では、ターゲット PreparedStatement に必要なことを実行できないか、低レベルの JDBC 作業を行う必要があります。Spring JDBC モジュールは、実行環境（ConnectionCallback や PreparedStatementCreator など）を構成し、パラメーター値（SqlParameterSource など）を操作するための API を提供します。StatementCallback などの低レベル操作の API にもアクセスできます。
Spring Integration 4.2 以降、MessagePreparedStatementSetter では、requestMessage コンテキストで、PreparedStatement のパラメーターを手動で指定できます。このクラスは、標準の Spring JDBC API の PreparedStatementSetter とまったく同じロールを果たします。実際には、JdbcMessageHandler が JdbcTemplate で execute を呼び出すときに、インライン PreparedStatementSetter 実装から直接呼び出されます。
この関数インターフェースオプションは sqlParameterSourceFactory と相互に排他的であり、requestMessage から PreparedStatement のパラメーターを取り込むためのより強力な代替手段として使用できます。例: File データを DataBase BLOB 列にストリーミング方式で保存する必要がある場合に役立ちます。次の例は、その方法を示しています。
@Bean
@ServiceActivator(inputChannel = "storeFileChannel")
public MessageHandler jdbcMessageHandler(DataSource dataSource) {
JdbcMessageHandler jdbcMessageHandler = new JdbcMessageHandler(dataSource,
"INSERT INTO imagedb (image_name, content, description) VALUES (?, ?, ?)");
jdbcMessageHandler.setPreparedStatementSetter((ps, m) -> {
ps.setString(1, m.getHeaders().get(FileHeaders.FILENAME));
try (FileInputStream inputStream = new FileInputStream((File) m.getPayload()); ) {
ps.setBlob(2, inputStream);
}
catch (Exception e) {
throw new MessageHandlingException(m, e);
}
ps.setClob(3, new StringReader(m.getHeaders().get("description", String.class)));
});
return jdbcMessageHandler;
}XML 構成の観点から、prepared-statement-setter 属性は <int-jdbc:outbound-channel-adapter> コンポーネントで利用可能です。MessagePreparedStatementSetter Bean 参照を指定できます。
バッチ更新
リクエストメッセージのペイロードが Iterable インスタンスの場合、バージョン 5.1 以降、JdbcMessageHandler は JdbcOperations.batchUpdate() を実行します。Iterable の各要素は、リクエストメッセージのヘッダーで Message にラップされます。通常の SqlParameterSourceFactory ベースの構成の場合、これらのメッセージは、前述の JdbcOperations.batchUpdate() 関数で使用される引数の SqlParameterSource[] を作成するために使用されます。MessagePreparedStatementSetter 構成が適用されると、BatchPreparedStatementSetter バリアントが各アイテムのそれらのメッセージを反復処理するために使用され、提供された MessagePreparedStatementSetter がそれらに対して呼び出されます。keysGenerated モードが選択されている場合、バッチ更新はサポートされません。
送信ゲートウェイ
送信ゲートウェイは、送信アダプターと受信アダプターの組み合わせに似ています。そのロールは、メッセージを処理し、それを使用して SQL クエリを実行し、応答チャネルに送信することで結果で応答することです。デフォルトでは、次の例に示すように、メッセージペイロードとヘッダーはクエリへの入力パラメーターとして使用できます。
<int-jdbc:outbound-gateway
update="insert into mythings (id, status, name) values (:headers[id], 0, :payload[thing])"
request-channel="input" reply-channel="output" data-source="dataSource" /> 上記の例の結果は、mythings テーブルにレコードを挿入し、影響を受けた行の数を示すメッセージ（ペイロードはマップ: {UPDATED=1}）を出力チャネルに返すことです。
更新クエリが自動生成キーの挿入である場合、keys-generated="true" を前述の例に追加することにより、生成されたキーを応答メッセージに追加できます（一部のデータベースプラットフォームではサポートされていないため、これはデフォルトではありません）。次の例は、変更された構成を示しています。
<int-jdbc:outbound-gateway
update="insert into mythings (status, name) values (0, :payload[thing])"
request-channel="input" reply-channel="output" data-source="dataSource"
keys-generated="true"/>次の例に示すように、更新カウントまたは生成されたキーの代わりに、選択クエリを提供して、結果（受信アダプターなど）から応答メッセージを実行および生成することもできます。
<int-jdbc:outbound-gateway
update="insert into foos (id, status, name) values (:headers[id], 0, :payload[foo])"
query="select * from foos where id=:headers[$id]"
request-channel="input" reply-channel="output" data-source="dataSource"/>Spring Integration 2.2 以降、更新 SQL クエリは必須ではなくなりました。query 属性または query 要素を使用して、選択クエリのみを提供できるようになりました。これは、一般的なゲートウェイやペイロードエンリッチャーなどを使用してデータをアクティブに取得する必要がある場合に非常に役立ちます。次に、結果から応答メッセージが生成され（受信アダプターの動作と同様）、応答チャネルに渡されます。次の例は、query 属性を使用する方法を示しています。
<int-jdbc:outbound-gateway
query="select * from foos where id=:headers[id]"
request-channel="input"
reply-channel="output"
data-source="dataSource"/> デフォルトでは、 |
チャネルアダプターと同様に、リクエストおよび応答用に SqlParameterSourceFactory インスタンスを提供することもできます。デフォルトは送信アダプターの場合と同じであるため、リクエストメッセージは式のルートとして使用できます。keys-generated="true" の場合、式のルートは生成されたキーです（1 つしかない場合はマップ、複数値の場合はマップのリスト）。
送信ゲートウェイには、DataSource または JdbcTemplate への参照が必要です。また、受信メッセージのクエリへのバインドを制御するために、SqlParameterSourceFactory を挿入することもできます。
バージョン 4.2 以降、request-prepared-statement-setter 属性は request-sql-parameter-source-factory の代替として <int-jdbc:outbound-gateway> で使用可能です。MessagePreparedStatementSetter Bean 参照を指定できます。これにより、実行前に、より洗練された PreparedStatement 準備が実装されます。
MessagePreparedStatementSetter の詳細については、送信チャネルアダプターを参照してください。
JDBC メッセージストア
Spring Integration は、2 つの JDBC 固有のメッセージストア実装を提供します。JdbcMessageStore は、アグリゲーターおよびクレームチェックパターンでの使用に適しています。JdbcChannelMessageStore 実装は、メッセージチャネル専用の、よりターゲットを絞ったスケーラブルな実装を提供します。
JdbcMessageStore を使用してメッセージチャネルをバックアップできることに注意してください。JdbcChannelMessageStore はその目的に最適化されています。
バージョン 5.0.11、5.1.2 以降、JdbcChannelMessageStore のインデックスが最適化されました。このようなストアに大きなメッセージグループがある場合は、インデックスを変更することができます。さらに、PriorityChannel のインデックスはコメント化されています。これは、JDBC によってサポートされているチャネルを使用しない限り、インデックスが必要ないためです。 |
OracleChannelMessageStoreQueryProvider を使用する場合、優先チャネルインデックスはクエリのヒントに含まれているため、追加する必要があります。 |
データベースの初期化
JDBC メッセージストアコンポーネントの使用を開始する前に、適切なオブジェクトでターゲットデータベースをプロビジョニングする必要があります。
Spring Integration には、データベースの初期化に使用できるサンプルスクリプトが付属しています。spring-integration-jdbc JAR ファイルでは、org.springframework.integration.jdbc パッケージにスクリプトがあります。さまざまな一般的なデータベースプラットフォーム用の作成スクリプトとドロップスクリプトの例を提供します。これらのスクリプトを使用する一般的な方法は、Spring JDBC データソース初期化子で参照することです。スクリプトはサンプルとして、また必要なテーブル名と列名の仕様として提供されていることに注意してください。本番用に拡張する必要がある場合があります（たとえば、インデックス宣言を追加することにより）。
汎用 JDBC メッセージストア
JDBC モジュールは、データベースに裏付けされた Spring Integration MessageStore （クレームチェックパターンで重要）および MessageGroupStore （アグリゲーターなどのステートフルパターンで重要）の実装を提供します。両方のインターフェースは JdbcMessageStore によって実装されており、次の例に示すように、XML でストアインスタンスを構成するためのサポートがあります。
<int-jdbc:message-store id="messageStore" data-source="dataSource"/>DataSource の代わりに JdbcTemplate を指定できます。
次の例は、他のいくつかのオプション属性を示しています。
<int-jdbc:message-store id="messageStore" data-source="dataSource"
lob-handler="lobHandler" table-prefix="MY_INT_"/> 前の例では、メッセージをラージオブジェクトとして扱うために LobHandler （Oracle で必要になることが多い）と、ストアによって生成されるクエリのテーブル名のプレフィックスを指定しました。テーブル名のプレフィックスは、デフォルトで INT_ になります。
バッキングメッセージチャネル
JDBC を使用してメッセージチャネルをバッキングする場合は、JdbcChannelMessageStore 実装を使用することをお勧めします。メッセージチャネルと組み合わせた場合にのみ機能します。
サポートされているデータベース
JdbcChannelMessageStore は、データベース固有の SQL クエリを使用して、データベースからメッセージを取得します。JdbcChannelMessageStore で ChannelMessageStoreQueryProvider プロパティを設定する必要があります。この channelMessageStoreQueryProvider は、指定した特定のデータベースの SQL クエリを提供します。Spring Integration は、次のリレーショナルデータベースのサポートを提供します。
PostgreSQL
HSQLDB
MySQL
Oracle
Derby
H2
SqlServer
Sybase
DB2
データベースがリストにない場合は、AbstractChannelMessageStoreQueryProvider クラスを継承して、独自のカスタムクエリを提供できます。
バージョン 4.0 は、メッセージが同じミリ秒で保存されている場合でも先入れ先出し（FIFO）キューを確保するために、MESSAGE_SEQUENCE 列をテーブルに追加しました。
カスタムメッセージ挿入
バージョン 5.0 以降、ChannelMessageStorePreparedStatementSetter クラスをオーバーロードすることにより、JdbcChannelMessageStore へのメッセージ挿入のカスタム実装を提供できます。これを使用して、異なる列を設定したり、テーブル構造や直列化戦略を変更したりできます。例: byte[] へのデフォルトの直列化の代わりに、その構造を JSON 文字列として保存できます。
次の例では、setValues のデフォルト実装を使用して共通の列を格納し、メッセージペイロードを varchar として格納する動作をオーバーライドします。
public class JsonPreparedStatementSetter extends ChannelMessageStorePreparedStatementSetter {
@Override
public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage,
Object groupId, String region, boolean priorityEnabled) throws SQLException {
// Populate common columns
super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
// Store message payload as varchar
preparedStatement.setString(6, requestMessage.getPayload().toString());
}
}通常、キューイングにリレーショナルデータベースを使用することはお勧めしません。代わりに、可能であれば、代わりに JMS または AMQP でバックアップされたチャネルの使用を検討してください。詳細については、次のリソースを参照してください。 |
同時ポーリング
メッセージチャネルをポーリングする場合、Poller を TaskExecutor 参照で構成するオプションがあります。
ただし、JDBC でバックアップされたメッセージチャネルを使用し、チャネルをポーリングして、その結果として複数のスレッドでメッセージストアをトランザクション的にポーリングする予定の場合は、マルチバージョン同時実行制御 [Wikipedia] (英語) (MVCC) をサポートするリレーショナルデータベースを使用する必要があります。そうしないと、ロックが課題になる可能性があり、複数のスレッドを使用する場合のパフォーマンスが期待どおりに実現されない可能性があります。たとえば、その点では Apache Derby に課題があります。 JDBC キューのスループットを向上させ、異なるスレッドがキューから同じ |
優先チャンネル
バージョン 4.0 以降、JdbcChannelMessageStore は PriorityCapableChannelMessageStore を実装し、priorityEnabled オプションを提供し、priority-queue インスタンスの message-store 参照として使用できるようにします。この目的のために、INT_CHANNEL_MESSAGE テーブルには PRIORITY メッセージヘッダーの値を格納する MESSAGE_PRIORITY 列があります。さらに、新しい MESSAGE_SEQUENCE 列により、同じ優先度で同じミリ秒に複数のメッセージが保存されている場合でも、堅牢な先入れ先出し（FIFO）ポーリングメカニズムを実現できます。メッセージは、order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE を使用してデータベースからポーリング（選択）されます。
priorityEnabled オプションはストア全体に適用され、適切な FIFO キューセマンティクスはキューチャネルに対して保持されないため、同じ JdbcChannelMessageStore Bean を優先キューチャネルと非優先キューチャネルに使用することはお勧めしません。ただし、同じ INT_CHANNEL_MESSAGE テーブル（および region も）を両方の JdbcChannelMessageStore 型に使用できます。そのシナリオを構成するには、次の例に示すように、一方のメッセージストア Bean を他方から拡張できます。 |
<bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
<property name="dataSource" ref="dataSource"/>
<property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
</bean>
<int:channel id="queueChannel">
<int:queue message-store="channelStore"/>
</int:channel>
<bean id="priorityStore" parent="channelStore">
<property name="priorityEnabled" value="true"/>
</bean>
<int:channel id="priorityChannel">
<int:priority-queue message-store="priorityStore"/>
</int:channel>メッセージストアのパーティション分割
JdbcMessageStore は、同じアプリケーション内のアプリケーションまたはノードのグループのグローバルストアとして使用するのが一般的です。名前の衝突に対する保護を提供し、データベースのメタデータ構成を制御できるようにするために、メッセージストアでは、2 つの方法でテーブルをパーティション分割できます。1 つの方法は、プレフィックスを変更することにより、個別のテーブル名を使用することです（前述のとおり）。もう 1 つの方法は、単一のテーブル内でデータをパーティション分割するために region 名を指定することです。2 番目のアプローチの重要な使用例は、MessageStore が Spring Integration メッセージチャネルをサポートする永続キューを管理している場合です。永続チャネルのメッセージデータは、チャネル名のストアにキー入力されます。その結果、チャネル名がグローバルに一意でない場合、チャネルはそれらに向けられていないデータを取得できます。この危険を回避するために、メッセージストア region を使用して、同じ論理名を持つ異なる物理チャネルのデータを別々に保つことができます。
ストアドプロシージャー
特定の状況では、プレーンな JDBC サポートでは不十分です。おそらく、従来のリレーショナルデータベーススキーマを扱うか、複雑なデータ処理のニーズがありますが、最終的には、ストアドプロシージャ [Wikipedia] (英語) またはストアドファンクションを使用する必要があります。Spring Integration 2.1 以降、ストアドプロシージャまたはストアドファンクションを実行する 3 つのコンポーネントを提供しています。
ストアドプロシージャの受信チャネルアダプター
ストアドプロシージャの送信チャネルアダプター
ストアドプロシージャの送信ゲートウェイ
サポートされているデータベース
ストアドプロシージャとストアドファンクションの呼び出しを可能にするために、ストアドプロシージャコンポーネントは org.springframework.jdbc.core.simple.SimpleJdbcCall (Javadoc) クラスを使用します。ストアドプロシージャの実行では、次のデータベースが完全にサポートされています。
Apache Derby
DB2
MySQL
Microsoft SQL Server
Oracle
PostgreSQL
Sybase
代わりにストアド関数を実行する場合、次のデータベースが完全にサポートされます。
MySQL
Microsoft SQL Server
Oracle
PostgreSQL
特定のデータベースが完全にサポートされていない場合でも、RDBMS がストアドプロシージャまたはストアド関数をサポートしていれば、ストアドプロシージャ Spring Integration コンポーネントを非常にうまく使用できる可能性があります。 実際のところ、提供されている統合テストのいくつかは H2 データベース (英語) を使用しています。それでも、これらの使用シナリオを徹底的にテストすることは非常に重要です。 |
共通の構成属性
すべてのストアドプロシージャコンポーネントは、特定の構成パラメーターを共有します。
auto-startup: アプリケーションコンテキストの起動時にこのコンポーネントを起動する必要があるかどうかを示すライフサイクル属性。デフォルトはtrueです。オプション。data-source: データベースへのアクセスに使用されるjavax.sql.DataSourceへの参照。必須。id: 発信チャネルアダプターのchannel属性がSubscribableChannelまたはPollableChannelのどちらを参照するかに応じて、基になる Spring Bean 定義を識別します。これは、EventDrivenConsumerまたはPollingConsumerのいずれかのインスタンスです。オプション。ignore-column-meta-data: 完全にサポートされているデータベースの場合、基になるSimpleJdbcCall(Javadoc) クラスは、JDBC メタデータからストアドプロシージャまたはストアドファンクションのパラメーター情報を自動的に取得できます。ただし、データベースがメタデータ検索をサポートしていない場合、またはカスタマイズされたパラメーター定義を提供する必要がある場合、このフラグを
trueに設定できます。デフォルトはfalseです。オプション。is-function:trueの場合、SQL 関数が呼び出されます。その場合、stored-procedure-nameまたはstored-procedure-name-expression属性は、呼び出された関数の名前を定義します。デフォルトはfalseです。オプション。stored-procedure-name: この属性は、ストアドプロシージャの名前を指定します。is-function属性がtrueに設定されている場合、この属性は代わりに関数名を指定します。このプロパティまたはstored-procedure-name-expressionのいずれかを指定する必要があります。stored-procedure-name-expression: この属性は、SpEL 式を使用してストアドプロシージャの名前を指定します。SpEL を使用すると、ヘッダーとペイロードを含む完全なメッセージ（利用可能な場合）にアクセスできます。この属性を使用して、実行時にさまざまなストアドプロシージャを呼び出すことができます。例: メッセージヘッダーとして実行するストアドプロシージャ名を指定できます。式はStringに解決する必要があります。is-function属性がtrueに設定されている場合、この属性はストアド関数を指定します。このプロパティまたはstored-procedure-nameのいずれかを指定する必要があります。jdbc-call-operations-cache-size: キャッシュされるSimpleJdbcCallOperationsインスタンスの最大数を定義します。基本的に、ストアドプロシージャ名ごとに、新しいSimpleJdbcCallOperations(Javadoc) インスタンスが作成され、そのインスタンスがキャッシュされます。Spring Integration 2.2 は、 stored-procedure-name-expression属性とjdbc-call-operations-cache-size属性を追加しました。デフォルトのキャッシュサイズは
10です。0の値はキャッシュを無効にします。負の値は許可されていません。JMX を有効にすると、
jdbc-call-operations-cacheに関する統計情報が MBean として公開されます。詳細については、MBean エクスポーターを参照してください。sql-parameter-source-factory: （ストアードプロシージャーの受信・チャネルアダプターには使用できません）SqlParameterSourceFactoryへの参照。デフォルトでは、渡されたMessageペイロードの Bean プロパティは、BeanPropertySqlParameterSourceFactoryを使用して、ストアドプロシージャの入力パラメーターのソースとして使用されます。これは基本的な使用例で十分です。より高度なオプションについては、1 つ以上の
ProcedureParameter値を渡すことを検討してください。パラメーターソースの定義を参照してください。オプション。use-payload-as-parameter-source: （ストアードプロシージャーの受信・チャネルアダプターでは使用できません）trueに設定されている場合、Messageのペイロードはパラメーターを提供するためのソースとして使用されます。ただし、falseに設定すると、Message全体がパラメーターのソースとして使用可能になります。プロシージャパラメーターが渡されない場合、このプロパティのデフォルトは
trueになります。これは、デフォルトのBeanPropertySqlParameterSourceFactoryを使用することにより、ペイロードの Bean プロパティがストアドプロシージャまたはストアドファンクションのパラメーター値のソースとして使用されることを意味します。ただし、プロシージャパラメーターが渡された場合、このプロパティは（デフォルトで）
falseに評価されます。ProcedureParameterでは、SpEL 式を提供できます。Message全体にアクセスできることは非常に有益です。このプロパティは、基になるStoredProcExecutorで設定されます。オプション。
共通の構成サブ要素
ストアドプロシージャコンポーネントは、パラメーターを定義してストアドプロシージャまたはストアド関数に渡すために使用できる子要素の共通セットを共有します。以下の要素が利用可能です:
parameterreturning-resultsetsql-parameter-definitionpollerparameter: ストアドプロシージャのパラメーターを提供するメカニズムを提供します。パラメーターは静的にすることも、SpEL 式を使用して提供することもできます。<int-jdbc:parameter name="" (1) type="" (2) value=""/> (3) <int-jdbc:parameter name="" expression=""/> (4)+ <1> ストアドプロシージャまたはストアド関数に渡されるパラメーターの名前。必須。<2> この属性は、値の型を指定します。何も指定されていない場合、この属性のデフォルトは
java.lang.Stringになります。この属性は、value属性が使用される場合にのみ使用されます。オプション。<3> パラメーターの値。この属性またはexpression属性のいずれかを提供する必要があります。オプション。<4>value属性の代わりに、パラメーターの値を渡すための SpEL 式を指定できます。expressionを指定した場合、value属性は許可されません。オプション。オプション。
returning-resultset: ストアドプロシージャは複数の結果セットを返す場合があります。1 つ以上のreturning-resultset要素を設定することにより、RowMappersを指定して、返された各ResultSetを意味のあるオブジェクトに変換できます。オプション。<int-jdbc:returning-resultset name="" row-mapper="" />sql-parameter-definition: 完全にサポートされているデータベースを使用する場合、通常、ストアドプロシージャのパラメーター定義を指定する必要はありません。代わりに、これらのパラメーターは JDBC メタデータから自動的に派生できます。ただし、完全にサポートされていないデータベースを使用する場合は、sql-parameter-definition要素を使用してこれらのパラメーターを明示的に設定する必要があります。ignore-column-meta-data属性を使用して、JDBC を介して取得したパラメーターメタデータ情報の処理をオフにすることもできます。<int-jdbc:sql-parameter-definition name="" (1) direction="IN" (2) type="STRING" (3) scale="5" (4) type-name="FOO_STRUCT" (5) return-type="fooSqlReturnType"/> (6)1 SQL パラメーターの名前を指定します。必須。 2 SQL パラメーター定義の方向を指定します。デフォルトは INです。有効な値は次のとおりです:IN、OUT、INOUT。プロシージャが結果セットを返す場合は、returning-resultset要素を使用してください。オプション。3 この SQL パラメーター定義に使用される SQL 型。 java.sql.Typesで定義されている整数値に変換します。または、整数値も提供できます。この属性が明示的に設定されていない場合、デフォルトは "VARCHAR" です。オプション。4 SQL パラメーターのスケール。数値および小数のパラメーターにのみ使用されます。オプション。 5 STRUCT、DISTINCT、JAVA_OBJECTなどのユーザー指定の型および名前付き配列型のtypeName。この属性は、scale属性と相互に排他的です。オプション。6 複合型のカスタム値ハンドラーへの参照。 SqlReturnType(Javadoc) の実装。この属性はscale属性と相互に排他的であり、OUT および INOUT パラメーターにのみ適用可能です。オプション。poller: このエンドポイントがPollingConsumerの場合、メッセージポーラーを構成できます。オプション。
パラメーターソースの定義
パラメーターソースは、Spring Integration メッセージプロパティを取得し、関連するストアドプロシージャの入力パラメーターにマッピングする手法を管理します。
ストアドプロシージャコンポーネントは、特定の規則に従います。デフォルトでは、Message ペイロードの Bean プロパティは、ストアドプロシージャの入力パラメーターのソースとして使用されます。その場合、BeanPropertySqlParameterSourceFactory が使用されます。これは基本的な使用例で十分です。次の例は、デフォルトの動作を示しています。
BeanPropertySqlParameterSourceFactory を使用した Bean プロパティの「自動」ルックアップを機能させるには、Bean プロパティを小文字で定義する必要があります。これは、org.springframework.jdbc.core.metadata.CallMetaDataContext （Java メソッドは matchInParameterValuesWithCallParameters()）では、取得されたストアドプロシージャのパラメーター宣言が小文字に変換されるためです。その結果、キャメルケース Bean プロパティ（lastName など）がある場合、検索は失敗します。その場合、明示的な ProcedureParameter を提供します。 |
次の 3 つのプロパティを持つ単純な Bean で構成されるペイロードがあるとします: id、name、description。さらに、id、name、description という 3 つの入力パラメーターを受け入れる INSERT_COFFEE と呼ばれる単純なストアドプロシージャがあります。また、完全にサポートされているデータベースを使用しています。その場合、ストアードプロシージャーの送信・アダプターの以下の構成で十分です。
<int-jdbc:stored-proc-outbound-channel-adapter data-source="dataSource"
channel="insertCoffeeProcedureRequestChannel"
stored-procedure-name="INSERT_COFFEE"/> より高度なオプションについては、1 つ以上の ProcedureParameter 値を渡すことを検討してください。
ProcedureParameter 値を明示的に指定した場合、デフォルトでは、ExpressionEvaluatingSqlParameterSourceFactory がパラメーター処理に使用され、SpEL 式のフルパワーが有効になります。
パラメーターの取得方法をさらに制御する必要がある場合は、sql-parameter-source-factory 属性を使用して SqlParameterSourceFactory のカスタム実装を渡すことを検討してください。
ストアドプロシージャ受信チャネルアダプター
次のリストは、ストアドプロシージャ受信チャネルアダプターに関係する属性を示しています。
<int-jdbc:stored-proc-inbound-channel-adapter
channel="" (1)
stored-procedure-name=""
data-source=""
auto-startup="true"
id=""
ignore-column-meta-data="false"
is-function="false"
skip-undeclared-results="" (2)
return-value-required="false" (3)
<int:poller/>
<int-jdbc:sql-parameter-definition name="" direction="IN"
type="STRING"
scale=""/>
<int-jdbc:parameter name="" type="" value=""/>
<int-jdbc:parameter name="" expression=""/>
<int-jdbc:returning-resultset name="" row-mapper="" />
</int-jdbc:stored-proc-inbound-channel-adapter>| 1 | ポーリングされたメッセージが送信されるチャネル。ストアドプロシージャまたは関数がデータを返さない場合、Message のペイロードは null です。必須。 |
| 2 | この属性が true に設定されている場合、対応する SqlOutParameter 宣言を持たないストアドプロシージャコールの結果はすべてバイパスされます。例: ストアドプロシージャが単一の結果パラメーターのみを宣言している場合でも、ストアドプロシージャは更新カウント値を返すことができます。正確な動作は、データベースの実装に依存します。値は、基礎となる JdbcTemplate に設定されます。値のデフォルトは true です。オプション。 |
| 3 | このプロシージャの戻り値を含めるかどうかを示します。Spring Integration 3.0 以降。オプション。 |
ストアドプロシージャの送信チャネルアダプター
次のリストは、ストアドプロシージャの送信チャネルアダプターに重要な属性を示しています。
<int-jdbc:stored-proc-outbound-channel-adapter channel="" (1)
stored-procedure-name=""
data-source=""
auto-startup="true"
id=""
ignore-column-meta-data="false"
order="" (2)
sql-parameter-source-factory=""
use-payload-as-parameter-source="">
<int:poller fixed-rate=""/>
<int-jdbc:sql-parameter-definition name=""/>
<int-jdbc:parameter name=""/>
</int-jdbc:stored-proc-outbound-channel-adapter>| 1 | このエンドポイントの受信メッセージチャネル。必須。 |
| 2 | このエンドポイントがチャンネルのサブスクライバーとして接続されている場合の呼び出しの順序を指定します。これは、そのチャネルが failover ディスパッチ戦略を使用している場合に特に重要です。このエンドポイント自体がキューを持つチャネルのポーリングコンシューマーである場合、効果はありません。オプション。 |
ストアドプロシージャの送信ゲートウェイ
次のリストは、ストアドプロシージャの送信チャネルアダプターに重要な属性を示しています。
<int-jdbc:stored-proc-outbound-gateway request-channel="" (1)
stored-procedure-name=""
data-source=""
auto-startup="true"
id=""
ignore-column-meta-data="false"
is-function="false"
order=""
reply-channel="" (2)
reply-timeout="" (3)
return-value-required="false" (4)
skip-undeclared-results="" (5)
sql-parameter-source-factory=""
use-payload-as-parameter-source="">
<int-jdbc:sql-parameter-definition name="" direction="IN"
type=""
scale="10"/>
<int-jdbc:sql-parameter-definition name=""/>
<int-jdbc:parameter name="" type="" value=""/>
<int-jdbc:parameter name="" expression=""/>
<int-jdbc:returning-resultset name="" row-mapper="" />| 1 | このエンドポイントの受信メッセージチャネル。必須。 |
| 2 | データベースレスポンスを受信した後、レスポンスが送信されるメッセージチャネル。オプション。 |
| 3 | このゲートウェイが、例外をスローする前に応答メッセージが正常に送信されるのを待つ時間を指定できます。DirectChannel に送信する場合、呼び出しは送信者のスレッドで発生することに注意してください。その結果、送信操作の失敗は、さらに下流の他のコンポーネントによって引き起こされる場合があります。デフォルトでは、ゲートウェイは無期限に待機します。値はミリ秒単位で指定されます。オプション。 |
| 4 | このプロシージャの戻り値を含めるかどうかを示します。オプション。 |
| 5 | skip-undeclared-results 属性が true に設定されている場合、対応する SqlOutParameter 宣言を持たないストアドプロシージャコールの結果はすべてバイパスされます。例: ストアドプロシージャが単一の結果パラメーターのみを宣言している場合でも、ストアドプロシージャは更新カウント値を返すことがあります。正確な動作はデータベースによって異なります。値は、基礎となる JdbcTemplate に設定されます。値のデフォルトは true です。オプション。 |
サンプル
このセクションには、Apache Derby (英語) ストアドプロシージャを呼び出す 2 つの例が含まれています。最初のプロシージャは、ResultSet を返すストアドプロシージャを呼び出します。RowMapper を使用することにより、データはドメインオブジェクトに変換され、その後、Spring Integration メッセージペイロードになります。
2 番目のサンプルでは、代わりに出力パラメーターを使用してデータを返すストアドプロシージャを呼び出します。
Spring Integration サンプルプロジェクト [GitHub] (英語) を参照してください。 プロジェクトには、ここで参照されている Apache Derby の例と、その実行方法の説明が含まれています。Spring Integration サンプルプロジェクトには、Oracle ストアドプロシージャの使用例 [GitHub] (英語) も用意されています。 |
最初の例では、FIND_ALL_COFFEE_BEVERAGES という名前のストアドプロシージャを呼び出します。このストアドプロシージャは、入力パラメーターを定義しませんが、ResultSet を返します。
Apache Derby では、ストアドプロシージャは Java で実装されています。次のリストはメソッドシグネチャーを示しています。
public static void findAllCoffeeBeverages(ResultSet[] coffeeBeverages)
throws SQLException {
...
}次のリストは、対応する SQL を示しています。
CREATE PROCEDURE FIND_ALL_COFFEE_BEVERAGES() \
PARAMETER STYLE JAVA LANGUAGE JAVA MODIFIES SQL DATA DYNAMIC RESULT SETS 1 \
EXTERNAL NAME 'o.s.i.jdbc.storedproc.derby.DerbyStoredProcedures.findAllCoffeeBeverages';Spring Integration では、次の例に示すように、たとえば stored-proc-outbound-gateway を使用して、このストアドプロシージャを呼び出すことができます。
<int-jdbc:stored-proc-outbound-gateway id="outbound-gateway-storedproc-find-all"
data-source="dataSource"
request-channel="findAllProcedureRequestChannel"
expect-single-result="true"
stored-procedure-name="FIND_ALL_COFFEE_BEVERAGES">
<int-jdbc:returning-resultset name="coffeeBeverages"
row-mapper="org.springframework.integration.support.CoffeBeverageMapper"/>
</int-jdbc:stored-proc-outbound-gateway>2 番目の例では、1 つの入力パラメーターを持つ FIND_COFFEE という名前のストアドプロシージャを呼び出します。ResultSet を返す代わりに、出力パラメーターを使用します。次の例は、メソッドのシグネチャーを示しています。
public static void findCoffee(int coffeeId, String[] coffeeDescription)
throws SQLException {
...
}次のリストは、対応する SQL を示しています。
CREATE PROCEDURE FIND_COFFEE(IN ID INTEGER, OUT COFFEE_DESCRIPTION VARCHAR(200)) \
PARAMETER STYLE JAVA LANGUAGE JAVA EXTERNAL NAME \
'org.springframework.integration.jdbc.storedproc.derby.DerbyStoredProcedures.findCoffee';Spring Integration では、次の例に示すように、たとえば stored-proc-outbound-gateway を使用して、このストアドプロシージャを呼び出すことができます。
<int-jdbc:stored-proc-outbound-gateway id="outbound-gateway-storedproc-find-coffee"
data-source="dataSource"
request-channel="findCoffeeProcedureRequestChannel"
skip-undeclared-results="true"
stored-procedure-name="FIND_COFFEE"
expect-single-result="true">
<int-jdbc:parameter name="ID" expression="payload" />
</int-jdbc:stored-proc-outbound-gateway>JDBC ロックレジストリ
バージョン 4.3 は JdbcLockRegistry を導入しました。特定のコンポーネント（アグリゲーターやリシーケンサーなど）は、LockRegistry インスタンスから取得したロックを使用して、一度に 1 つのスレッドのみがグループを操作するようにします。DefaultLockRegistry は、単一のコンポーネント内でこの機能を実行します。これらのコンポーネントで外部ロックレジストリを構成できるようになりました。共有 MessageGroupStore で使用する場合、JdbcLockRegistry を使用して複数のアプリケーションインスタンスにこの機能を提供し、一度に 1 つのインスタンスのみがグループを操作できます。
ローカルスレッドによってロックが解除されると、通常、別のローカルスレッドがすぐにロックを取得できます。別のレジストリインスタンスを使用するスレッドによってロックが解放された場合、ロックを取得するのに最大 100 ミリ秒かかることがあります。
JdbcLockRegistry は LockRepository 抽象化に基づいており、DefaultLockRepository 実装があります。データベーススキーマスクリプトは org.springframework.integration.jdbc パッケージにあり、特定の RDBMS ベンダー用に分割されています。例: 次のリストは、ロックテーブルの H2 DDL を示しています。
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint INT_LOCK_PK primary key (LOCK_KEY, REGION)
);INT_ は、ターゲットデータベースの設計要件に応じて変更できます。DefaultLockRepository Bean 定義で prefix プロパティを使用する必要があります。
あるアプリケーションが、分散ロックを解除できず、データベース内の特定のレコードを削除できない状態に移行する場合があります。この目的のために、そのようなデッドロックは、次のロック呼び出しで他のアプリケーションによって期限切れになる可能性があります。DefaultLockRepository の timeToLive （TTL）オプションは、この目的のために提供されています。また、特定の DefaultLockRepository インスタンスに保存されているロックに CLIENT_ID を指定することもできます。その場合、id をコンストラクターパラメーターとして DefaultLockRepository に関連付けるように指定できます。
バージョン 5.1.8 から、JdbcLockRegistry は idleBetweenTries （ロックレコードの挿入 / 更新の実行間でスリープする Duration）で構成できます。デフォルトでは 100 ミリ秒であり、一部の環境ではリーダー以外のユーザーがデータソースとの接続を頻繁に汚染します。
バージョン 5.4 以降、RenewableLockRegistry インターフェースが導入され、JdbcLockRegistry に追加されました。ロックされたプロセスがロックの存続時間よりも長くなる場合は、ロックされたプロセス中に renewLock() メソッドを呼び出す必要があります。そのため、存続時間を大幅に短縮でき、デプロイは失われたロックをすばやく取り戻すことができます。
| ロックの更新は、ロックが現在のスレッドによって保持されている場合にのみ実行できます。 |
JDBC メタデータストア
バージョン 5.0 は、JDBC MetadataStore （メタデータストアを参照）実装を導入しました。JdbcMetadataStore を使用して、アプリケーションの再起動後もメタデータの状態を維持できます。この MetadataStore 実装は、次のようなアダプターで使用できます。
これらのアダプターを構成して JdbcMetadataStore を使用するには、Bean の名前 metadataStore を使用して Spring Bean を宣言します。次の例に示すように、フィード受信チャネルアダプターとフィード受信チャネルアダプターは両方とも、宣言された JdbcMetadataStore を自動的に取得して使用します。
@Bean
public MetadataStore metadataStore(DataSource dataSource) {
return new JdbcMetadataStore(dataSource);
}org.springframework.integration.jdbc パッケージには、いくつかの RDMBS ベンダー用のデータベーススキーマスクリプトが含まれています。例: 次のリストは、メタデータテーブルの H2 DDL を示しています。
CREATE TABLE INT_METADATA_STORE (
METADATA_KEY VARCHAR(255) NOT NULL,
METADATA_VALUE VARCHAR(4000),
REGION VARCHAR(100) NOT NULL,
constraint INT_METADATA_STORE_PK primary key (METADATA_KEY, REGION)
);INT_ プレフィックスを変更して、ターゲットデータベースの設計要件に一致させることができます。カスタムプレフィックスを使用するように JdbcMetadataStore を構成することもできます。
JdbcMetadataStore は ConcurrentMetadataStore を実装し、複数のアプリケーションインスタンス間で確実に共有できるようにします。1 つのインスタンスのみがキーの値を保存または変更できます。トランザクションの保証により、これらの操作はすべてアトミックです。
トランザクション管理では JdbcMetadataStore を使用する必要があります。受信チャネルアダプターには、ポーラー構成の TransactionManager への参照を提供できます。JdbcMetadataStore を使用した非トランザクション MetadataStore 実装とは異なり、エントリは、トランザクションがコミットされた後にのみターゲットテーブルに表示されます。ロールバックが発生すると、INT_METADATA_STORE テーブルにエントリは追加されません。
バージョン 5.0.7 以降、メタデータストアエントリのロックベースのクエリ用に RDBMS ベンダー固有の lockHint オプションを使用して JdbcMetadataStore を構成できます。デフォルトでは、FOR UPDATE であり、ターゲットデータベースが行ロック機能をサポートしていない場合は、空の文字列で構成できます。更新前に行をロックするための SELECT 式の特定の可能なヒントについては、ベンダーに相談してください。