TransactionalPollerはSpringIntegrationです

私はメッセージストア属性を持つQueueChannelを持っています。その構成TransactionalPollerはSpringIntegrationです

<int:channel id="channel1"> 
    <int:queue message-store="msgStore"></int:queue 
</int:channel> 

<int-jdbc:message-store id="msgStore" data-source="jdbcSQLiteDataSource"/>

次に、httpプロトコルを使用してこのチャネルから別のアプリケーションにメッセージを送信します。私の設定：

<int-http:outbound-gateway 
    url="someUrl" http-method="PUT" 
    request-channel="channel1"> 

     <int-http:request-handler-advice-chain> 
     <int:retry-advice max-attempts="3"> 
      <int:exponential-back-off initial="1000" multiplier="2.0" maximum="8000"/> 
     </int:retry-advice> 
     </int-http:request-handler-advice-chain> 

</int-http:outbound-gateway>

を受信側アプリケーションは、私のメッセージを取得していない場合（例えば、私が受けたホストからタイムアウト例外を取得）私はバックチャンネル（チャンネル1）で出て、このメッセージを返すようにしたいです。

私はトランザクションポーラを使用しますが、動作しません。メッセージはチャンネルから削除されます。

<int:poller default="true" fixed-rate="100" receive-timeout="100"> 
    <int:transactional transaction-manager="transactionManager"/> 
</int:poller> 

<bean id="transactionManager" 
     class="org.springframework.jdbc.datasource.DataSourceTransactionManager"> 
    <property name="dataSource" ref="jdbcSQLiteDataSource"/> 
</bean>

LOG

DEBUG o.s.i.channel.QueueChannel: preSend on channel 'channel1', message: GenericMessage [payload=1, headers={timestamp=1481896454785, id=90adf115-d76e-3066-de99-ff3b867b66a5}] 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL query 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL statement [SELECT COUNT(GROUP_KEY) FROM INT_MESSAGE_GROUP where GROUP_KEY=? and REGION=?] 
    DEBUG o.s.j.datasource.DataSourceUtils: Fetching JDBC Connection from DataSource 
    INFO o.s.j.datasource.SingleConnectionDataSource: Established shared JDBC Connection: [email protected] 


    DEBUG o.s.j.datasource.DataSourceTransactionManager: Acquired Connection [[email protected]] for JDBC transaction 

    DEBUG o.s.j.datasource.DataSourceUtils: Changing isolation level of JDBC Connection [[email protected]] to 8 
    DEBUG o.s.i.jdbc.JdbcMessageStore: Removing messages from group with group key=c5404fa9-4717-38a5-8871-b8761a8414d5 
    DEBUG o.s.j.core.JdbcTemplate: Executing SQL batch update [DELETE from INT_GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=? and REGION=?] with a batch size of 100 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL statement [DELETE from INT_GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=? and REGION=?] 
    DEBUG o.s.j.support.JdbcUtils: JDBC driver supports batch updates 
    DEBUG o.s.j.core.JdbcTemplate: Sending SQL batch update #1 with 1 items 


    DEBUG o.s.j.core.JdbcTemplate: Executing SQL batch update [DELETE from INT_MESSAGE where MESSAGE_ID=? and REGION=?] with a batch size of 100 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL statement [**DELETE from INT_MESSAGE where MESSAGE_ID=? and REGION=?]** 
    DEBUG o.s.j.support.JdbcUtils: JDBC driver supports batch updates 
    DEBUG o.s.j.core.JdbcTemplate: Sending SQL batch update #1 with 1 items 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL update 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL statement [UPDATE INT_MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?] 
    DEBUG o.s.i.jdbc.JdbcMessageStore: Updating MessageGroup: c5404fa9-4717-38a5-8871-b8761a8414d5 
    DEBUG o.s.j.core.JdbcTemplate: SQL update affected 1 rows 
    DEBUG o.s.i.channel.QueueChannel: postReceive on channel 'channel1', message: GenericMessage [payload=1, headers={id=90adf115-d76e-3066-de99-ff3b867b66a5, timestamp=1481896454898, JdbcMessageStore.CREATED_DATE=1481896454898, JdbcMessageStore.SAVED=true}] 
    DEBUG o.s.i.endpoint.PollingConsumer: Poll resulted in Message: GenericMessage [payload=1, headers={id=90adf115-d76e-3066-de99-ff3b867b66a5, timestamp=1481896454898, JdbcMessageStore.CREATED_DATE=1481896454898, JdbcMessageStore.SAVED=true}] 
    DEBUG o.s.i.http.outbound.HttpRequestExecutingMessageHandler: o.s.i.http.outbound.HttpRequestExecutingMessageHandler#0 received message: GenericMessage [payload=1, headers={id=90adf115-d76e-3066-de99-ff3b867b66a5, timestamp=1481896454898, JdbcMessageStore.CREATED_DATE=1481896454898, JdbcMessageStore.SAVED=true}] 
    DEBUG org.springframework.retry.support.RetryTemplate: Retry: count=0 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: outboundHeaderNames=[Accept, Accept-Charset, Accept-Encoding, Accept-Language, Accept-Ranges, Authorization, Cache-Control, Connection, Content-Length, Content-Type, Cookie, Date, Expect, From, Host, If-Match, If-Modified-Since, If-None-Match, If-Range, If-Unmodified-Since, Max-Forwards, Pragma, Proxy-Authorization, Range, Referer, TE, Upgrade, User-Agent, Via, Warning] 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[id] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[timestamp] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[jdbcmessagestore.created_date] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[jdbcmessagestore.saved] WILL NOT be mapped 

    DEBUG org.springframework.web.client.RestTemplate: **Created PUT request for** "http://localhost:8081/test/... 

    DEBUG org.springframework.web.client.RestTemplate: Writing [1] as "text/plain;charset=UTF-8" using [[email protected]56e60b] 
    DEBUG org.springframework.retry.backoff.ExponentialBackOffPolicy: Sleeping for 1000 
    DEBUG org.springframework.retry.support.RetryTemplate: Checking for rethrow: count=1 
    DEBUG org.springframework.retry.support.RetryTemplate: Retry: count=1 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: outboundHeaderNames=[Accept, Accept-Charset, Accept-Encoding, Accept-Language, Accept-Ranges, Authorization, Cache-Control, Connection, Content-Length, Content-Type, Cookie, Date, Expect, From, Host, If-Match, If-Modified-Since, If-None-Match, If-Range, If-Unmodified-Since, Max-Forwards, Pragma, Proxy-Authorization, Range, Referer, TE, Upgrade, User-Agent, Via, Warning] 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[id] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[timestamp] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[jdbcmessagestore.created_date] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[jdbcmessagestore.saved] WILL NOT be mapped 


    DEBUG org.springframework.web.client.RestTemplate: **Created PUT request** for "http://localhost:8081/test/... 


    DEBUG org.springframework.web.client.RestTemplate: Writing [1] as "text/plain;charset=UTF-8" using [[email protected]56e60b] 
    DEBUG org.springframework.retry.backoff.ExponentialBackOffPolicy: Sleeping for 2000 
    DEBUG org.springframework.retry.support.RetryTemplate: Checking for rethrow: count=2 
    DEBUG org.springframework.retry.support.RetryTemplate: Retry: count=2 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: outboundHeaderNames=[Accept, Accept-Charset, Accept-Encoding, Accept-Language, Accept-Ranges, Authorization, Cache-Control, Connection, Content-Length, Content-Type, Cookie, Date, Expect, From, Host, If-Match, If-Modified-Since, If-None-Match, If-Range, If-Unmodified-Since, Max-Forwards, Pragma, Proxy-Authorization, Range, Referer, TE, Upgrade, User-Agent, Via, Warning] 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[id] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[timestamp] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[jdbcmessagestore.created_date] WILL NOT be mapped 
    DEBUG o.s.i.http.support.DefaultHttpHeaderMapper: headerName=[jdbcmessagestore.saved] WILL NOT be mapped 


    DEBUG org.springframework.web.client.RestTemplate: **Created PUT request** for "http://localhost:8081/test/... 


    DEBUG org.springframework.web.client.RestTemplate: Writing [1] as "text/plain;charset=UTF-8" using [[email protected]56e60b] 
    DEBUG org.springframework.retry.support.RetryTemplate: Checking for rethrow: count=3 
    DEBUG org.springframework.retry.support.RetryTemplate: Retry failed last attempt: count=3 
    DEBUG o.s.j.datasource.DataSourceTransactionManager: Initiating transaction rollback 
    DEBUG o.s.j.datasource.DataSourceTransactionManager: Rolling back JDBC transaction on Connection [[email protected]] 
    DEBUG o.s.j.datasource.DataSourceTransactionManager: Releasing JDBC Connection [[email protected]] after transaction 
    DEBUG o.s.j.datasource.DataSourceUtils: Returning JDBC Connection to DataSource 
    DEBUG o.s.b.factory.support.DefaultListableBeanFactory: Returning cached instance of singleton bean 'errorChannel' 
    DEBUG o.s.i.channel.PublishSubscribeChannel: preSend on channel 'errorChannel', message: ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: HTTP request execution failed for URI [http://localhost:8081/test/...]; nested exception is org.springframework.web.client.ResourceAccessException: I/O error on PUT request for "http://localhost:8081/test/...": Connection refused: connect; nested exception is java.net.ConnectException: Connection refused: connect, headers={timestamp=1481896460962, id=8bff6f2e-ef3a-52bd-5d41-f8785b45a880}] 
    DEBUG o.s.i.handler.LoggingHandler: _o.s.i.errorLogger.handler received message: ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: HTTP request execution failed for URI [http://localhost:8081/test/...]; nested exception is org.springframework.web.client.ResourceAccessException: I/O error on PUT request for "http://localhost:8081/test/...": Connection refused: connect; nested exception is java.net.ConnectException: Connection refused: connect, headers={timestamp=1481896460962, id=8bff6f2e-ef3a-52bd-5d41-f8785b45a880}] 
    ERROR o.s.i.handler.LoggingHandler: org.springframework.messaging.MessageHandlingException: HTTP request execution failed for URI [http://localhost:8081/test/...]; nested exception is org.springframework.web.client.ResourceAccessException: I/O error on PUT request for "http://localhost:8081/test/...": Connection refused: connect; nested exception is java.net.ConnectException: Connection refused: connect 
Caused by: java.net.ConnectException: Connection refused: connect  
... 

    DEBUG o.s.i.channel.PublishSubscribeChannel: postSend (sent=true) on channel 'errorChannel', message: ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: HTTP request execution failed for URI [http://localhost:8081/test/...]; nested exception is org.springframework.web.client.ResourceAccessException: I/O error on PUT request for "http://localhost:8081/test/...": Connection refused: connect; nested exception is java.net.ConnectException: Connection refused: connect, headers={timestamp=1481896460962, id=8bff6f2e-ef3a-52bd-5d41-f8785b45a880}] 
    DEBUG o.s.j.datasource.DataSourceTransactionManager: Creating new transaction with name [o.s.i.endpoint.AbstractPollingEndpoint$1.call]: PROPAGATION_REQUIRED,ISOLATION_SERIALIZABLE 


    DEBUG o.s.j.datasource.DataSourceTransactionManager: Acquired Connection [[email protected]] for JDBC transaction 
    DEBUG o.s.j.datasource.DataSourceUtils: Changing isolation level of JDBC Connection [[email protected]] to 8 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL query 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL statement [SELECT COUNT(MESSAGE_ID) from INT_GROUP_TO_MESSAGE where GROUP_KEY=? and REGION=?] 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL query 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL statement [SELECT COUNT(MESSAGE_ID) from INT_GROUP_TO_MESSAGE where GROUP_KEY=? and REGION=?] 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL query 
    DEBUG o.s.j.core.JdbcTemplate: Executing prepared SQL statement [SELECT INT_MESSAGE.MESSAGE_ID, INT_MESSAGE.MESSAGE_BYTES from INT_MESSAGE where INT_MESSAGE.MESSAGE_ID = (SELECT min(m.MESSAGE_ID) from INT_MESSAGE m join INT_GROUP_TO_MESSAGE on m.MESSAGE_ID = INT_GROUP_TO_MESSAGE.MESSAGE_ID where CREATED_DATE = (SELECT min(CREATED_DATE) from INT_MESSAGE, INT_GROUP_TO_MESSAGE where INT_MESSAGE.MESSAGE_ID = INT_GROUP_TO_MESSAGE.MESSAGE_ID and INT_GROUP_TO_MESSAGE.GROUP_KEY = ? and INT_MESSAGE.REGION = ?) and INT_GROUP_TO_MESSAGE.GROUP_KEY = ? and m.REGION = ?)] 
    DEBUG o.s.i.endpoint.PollingConsumer: Received no Message during the poll, returning 'false' 
    DEBUG o.s.j.datasource.DataSourceTransactionManager: Initiating transaction commit

私は何をしているのですが間違っているのですか？

出典

2016-12-16 Konstantin B.

他の質問への回答で提案したように、ログを見ましたか？ @GaryRussell。 –

はい、あります。私はSQLの削除がHTTP要求の前に実行されることがわかります。私の編集した投稿を参照してください –

多分あなたのデータベースです。私はSQLLiteに精通していないです私はH2でそれをテストし、すべてが期待...

<int:channel id="channel1"> 
    <int:queue message-store="msgStore" /> 
</int:channel> 

<int-jdbc:message-store id="msgStore" 
    data-source="dataSource" /> 

<int:service-activator input-channel="channel1" expression="1/0"> 
    <int:request-handler-advice-chain> 
     <int:retry-advice> 
      <int:fixed-back-off interval="1000" /> 
     </int:retry-advice> 
    </int:request-handler-advice-chain> 
</int:service-activator> 

<int:poller default="true" fixed-rate="100" receive-timeout="100"> 
    <int:transactional transaction-manager="transactionManager" /> 
</int:poller> 

<jdbc:initialize-database> 
    <jdbc:script location="classpath:org/springframework/integration/jdbc/schema-h2.sql" /> 
</jdbc:initialize-database> 


@SpringBootApplication 
@ImportResource("context.xml") 
public class So41184992Application { 

    public static void main(String[] args) throws Exception { 
     ConfigurableApplicationContext context = SpringApplication.run(So41184992Application.class, args); 
     context.getBean("channel1", MessageChannel.class).send(new GenericMessage<>("foo")); 
     Thread.sleep(30000); 
     context.close(); 
    } 

}

が結果として動作します。例外は2秒ごとに（1秒遅れで3回の試行を発生して

10:22:42.680 [task-scheduler-1] ERROR o.s.i.handler.LoggingHandler - org.springframework.messaging.MessageHandlingException: Expression evaluation failed: 1/0; nested exception is java.lang.ArithmeticException:/by zero 
at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:143) 
... 
10:22:44.690 [task-scheduler-1] ERROR o.s.i.handler.LoggingHandler - org.springframework.messaging.MessageHandlingException: Expression evaluation failed: 1/0; nested exception is java.lang.ArithmeticException:/by zero 
at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:143)

）。

ところでEDIT

、それは一般的なメッセージストアではなく、チャネルをバックアップするJdbcChannelMessageStoreを使用することが一般的に良好である - （凝集等のためのメッセージグループとは対照的に）それはチャネルのために最適化されます。

<bean id="msgStore" class="org.springframework.integration.jdbc.store.JdbcChannelMessageStore"> 
    <property name="dataSource" ref="dataSource" /> 
    <property name="channelMessageStoreQueryProvider"> 
     <bean class="org.springframework.integration.jdbc.store.channel.H2ChannelMessageStoreQueryProvider" /> 
    </property> 
</bean> 

<jdbc:initialize-database> 
    <jdbc:script location="classpath:org/springframework/integration/jdbc/store/channel/schema-h2.sql" /> 
</jdbc:initialize-database>

ただし、非標準データベースのカスタムクエリプロバイダを提供する必要があります。

出典

2016-12-16 15:29:04

私の編集も見てください。 –

あなたは私を理解できないかもしれません。参照してください。属性max-attempts = 3を設定します。 3回失敗した後、このメッセージをチャンネルに戻したいと思います。出来ますか？ –

これは私のテストケースがするものです。デフォルトの 'maxAttempts'は3です。 –

TransactionalPollerはSpringIntegrationです

答えて

関連する問題