TransactionalPoller ist SpringIntegration

Ich habe QueueChannel mit dem Message-Store-Attribut. Seine KonfigurationTransactionalPoller ist SpringIntegration

Dann sende ich Nachrichten von diesem Kanal zu einer anderen Anwendung mit HTTP-Protokoll. Meine Konfiguration:

<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>

Wenn die empfangende Anwendung nicht meine Botschaft nicht erhalten (zum Beispiel ich die Timeout Ausnahme von dem empfangenden Host bekommen) Ich möchte aus dieser Nachricht, um wieder zurück in dem Kanal (channel1).

Ich benutze einen Transaktions-Poller, aber es funktioniert nicht. Nachrichten werden aus dem Kanal entfernt.

<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

Was soll ich tun, ist falsch?

Quelle

2016-12-16 Konstantin B.

Haben Sie sich die Protokolle angesehen, wie ich in der Antwort auf Ihre andere Frage vorgeschlagen habe? –

@ GaryRussell. Ja, habe ich. Ich sehe, dass SQL-Löschung vor der HTTP-Anfrage läuft. Bitte, siehe meinen bearbeiteten Post –

Vielleicht ist es Ihre Datenbank; Ich bin nicht vertraut mit SQLLite Getestet habe ich es nur mit H2 und alles funktioniert wie erwartet ...

<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(); 
    } 

}

Ergebnis:

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)

Mit den Ausnahmen alle 2 Sekunden (3 Versuche mit 1 Sekunde Verzögerung auftritt,).

EDIT

By the way, ist es im Allgemeinen besser, ein JdbcChannelMessageStore zu verwenden Kanäle zu unterstützen, anstatt ein allgemeiner Nachrichtenspeicher - es für die Kanäle optimiert ist (wie etc für die Aggregation zu Nachrichtengruppen entgegengesetzt).

<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>

Sie müssten jedoch einen benutzerdefinierten Abfrage-Provider für eine nicht standardmäßige Datenbank bereitstellen.

Quelle

2016-12-16 15:29:04

Siehe auch meine Bearbeitung. –

Sie können mich nicht verstehen. Siehe, ich setze das Attribut max-versuche = 3. Ich möchte diese Nachricht nach 3 fehlgeschlagenen Versuchen wieder im Kanal ausgeben. Ist es möglich? –

Das ist genau das, was mein Testfall tut - der Standard 'maxAttempts' ist 3. –

TransactionalPoller ist SpringIntegration

Antwort

Verwandte Themen