为什么重试策略不适用于 spring 集成拆分器？
Why Retry Policy is not working for spring integration splitter?
重试政策无效
<task:executor id="ticketSourceExc" pool-size="1"
queue-capacity="0" rejection-policy="CALLER_RUNS" />
<task:executor id="regulatoryExc" pool-size="1"
queue-capacity="5" rejection-policy="CALLER_RUNS" />
<int:service-activator input-channel="ticketCacheChannel"
output-channel="sourceTicketsSplitter" ref="ticketSerActivator"
method="fetchDataFromDB">
<int:poller fixed-rate="10" task-executor="ticketSourceExc" />
</int:service-activator>
<!-- I am getting List<Tickets> -->
<int:splitter input-channel="sourceTicketsSplitter"
output-channel="sourceTicketChannel">
<int:request-handler-advice-chain>
<ref bean="retrier" />
</int:request-handler-advice-chain>
<int:poller fixed-rate="10" />
</int:splitter>
<int:handler-retry-advice id="retrier" max-attempts="4" recovery-channel="myErrorChannel" >
<int:exponential-back-off initial="1000" multiplier="5" maximum="6000"/>
</int:handler-retry-advice>
<int:bridge id="regulatoryBrigde" input-channel="sourceTicketChannel"
output-channel="regulatoryChannel">
<int:poller fixed-rate="10" task-executor="regulatoryExc" />
</int:bridge>
<int:chain id="regulatoryChainFlow" input-channel="regulatoryChannel">
<int:service-activator ref="regulatoryTaskActivator1"
method="process" />
<int:service-activator ref="regulatoryTaskActivator2"
method="process" />
<int:service-activator ref="regulatoryTaskActivator3"
method="process" />
</int:chain>
我已经为拆分器配置重试策略。输出通道是深度为 1 的队列通道（用于测试目的）。我的期望是当我得到一个包含 10 个项目的列表时，它应该在队列 full.It 之后进入异常通道没有发生 way.I 已经将线程睡眠放在 regulatoryTaskActivator1 中以阻止队列
这是正确的，因为发送到输出通道不包括在通知中。这个仅用于 request 部分。请仔细阅读有关问题的文档：https://docs.spring.io/spring-integration/docs/5.0.8.RELEASE/reference/html/messaging-endpoints-chapter.html#message-handler-advice-chain。该建议仅适用于 handleRequestMessage() 方法。
更新
对于您想要处理有限大小的队列并通过重试进行一些错误处理的用例，我建议您将拆分结果输出到 service-activator 周围 @MessagingGateway 在 @Gateway 方法上使用适当的 error-channel 和可能的 @Retryable：
<splitter input-channel="sourceTicketsSplitter"
output-channel="gatewayInputChannel">
<poller fixed-rate="10" />
</splitter>
<service-activator input-channel="gatewayInputChannel" ref="gateway">
<request-handler-advice-chain>
<ref bean="retrier" />
</request-handler-advice-chain>
</service-activator>
<gateway id="gateway" default-request-channel="sourceTicketChannel"/>
如果您的流程是 单向 并且您不希望收到 regulatoryChainFlow 的任何回复，则默认 RequestReplyExchanger 不适合您的要求，您需要使用 void 方法为网关引入一个简单的接口，并将 <gateway> 为此配置为 service-interface.
更新
我忘了告诉你的是，需要使用timeout发送到受限队列。如果没有这样的超时，发件人只需坐下来等待房间，该房间出现在提到的 Thread.sleep(10000);.
之后
那么，你需要的是这样的：
<int:gateway id="stateGateWay" default-request-channel="stateChannel"
default-request-timeout="100"
service-interface="com.biswo.myspringapp.gateway.StateGateWay" />
关注：
<xsd:attribute name="default-request-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
Provides the amount of time dispatcher would wait to send a message.
This timeout would only apply if there is a potential to block in the send call.
For example if this gateway is hooked up to a Queue channel.
Value is specified in milliseconds; it can be a simple long value or a SpEL
expression; array variable #args is available.
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
在此之后我开始在日志中看到 reties 然后 Dispatcher has no subscribers:
2018-10-09 14:53:03.466 TRACE 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : RetryContext retrieved: [RetryContext: count=0, lastException=null, exhausted=false]
2018-10-09 14:53:03.466 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry: count=0
2018-10-09 14:53:03.568 DEBUG 15808 --- [ask-scheduler-3] o.s.r.backoff.ExponentialBackOffPolicy : Sleeping for 100
2018-10-09 14:53:03.669 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Checking for rethrow: count=1
2018-10-09 14:53:03.669 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry: count=1
2018-10-09 14:53:03.771 DEBUG 15808 --- [ask-scheduler-3] o.s.r.backoff.ExponentialBackOffPolicy : Sleeping for 500
2018-10-09 14:53:04.271 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Checking for rethrow: count=2
2018-10-09 14:53:04.271 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry: count=2
2018-10-09 14:53:04.372 DEBUG 15808 --- [ask-scheduler-3] o.s.r.backoff.ExponentialBackOffPolicy : Sleeping for 600
2018-10-09 14:53:04.973 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Checking for rethrow: count=3
2018-10-09 14:53:04.973 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry: count=3
2018-10-09 14:53:05.073 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Checking for rethrow: count=4
2018-10-09 14:53:05.073 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry failed last attempt: count=4
2018-10-09 14:53:05.077 WARN 15808 --- [ask-scheduler-3] o.s.i.c.MessagePublishingErrorHandler : Error message was not delivered.
org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.errorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.errorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.messaging.MessageDeliveryException: Failed to send message to channel 'stateChannel' within timeout: 100, failedMessage=GenericMessage [payload=State [id=2426, name=Pamanzi, countryId=141], headers={id=710d13df-dfbb-ba95-8d28-4785d1a1e3d3, timestamp=1539111184973}], failedMessage=GenericMessage [payload=State [id=2426, name=Pamanzi, countryId=141], headers={COUNTRY_ID=141, sequenceNumber=2, COUNTRY=com.biswo.myspringapp.model.Country@5fd4ea2d, sequenceSize=2, correlationId=ab423aa4-c80e-6af0-33a8-f57337f25352, id=a70d6b11-fc23-edd9-7973-1342c3fab8a6, timestamp=1539111183466}], headers={id=ee0ffd29-8e6a-ab5a-6ffb-1016fa38457e, timestamp=1539111185074}], failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.messaging.MessageDeliveryException: Failed to send message to channel 'stateChannel' within timeout: 100, failedMessage=GenericMessage [payload=State [id=2426, name=Pamanzi, countryId=141], headers={id=710d13df-dfbb-ba95-8d28-4785d1a1e3d3, timestamp=1539111184973}], failedMessage=GenericMessage [payload=State [id=2426, name=Pamanzi, countryId=141], headers={COUNTRY_ID=141, sequenceNumber=2, COUNTRY=com.biswo.myspringapp.model.Country@5fd4ea2d, sequenceSize=2, correlationId=ab423aa4-c80e-6af0-33a8-f57337f25352, id=a70d6b11-fc23-edd9-7973-1342c3fab8a6, timestamp=1539111183466}], headers={id=ee0ffd29-8e6a-ab5a-6ffb-1016fa38457e, timestamp=1539111185074}], headers={id=34d0e6e6-0af8-1078-5dd7-aa6643ba6c8d, timestamp=1539111185075}] for original GenericMessage [payload=[State [id=2425, name=Mayotte, countryId=141], State [id=2426, name=Pamanzi, countryId=141]], headers={COUNTRY_ID=141, id=ab423aa4-c80e-6af0-33a8-f57337f25352, COUNTRY=com.biswo.myspringapp.model.Country@5fd4ea2d, timestamp=1539111183462}]
仅仅因为您的 errorChannel 是 DirectChannel 并且没有任何消费者。
重试政策无效
<task:executor id="ticketSourceExc" pool-size="1"
queue-capacity="0" rejection-policy="CALLER_RUNS" />
<task:executor id="regulatoryExc" pool-size="1"
queue-capacity="5" rejection-policy="CALLER_RUNS" />
<int:service-activator input-channel="ticketCacheChannel"
output-channel="sourceTicketsSplitter" ref="ticketSerActivator"
method="fetchDataFromDB">
<int:poller fixed-rate="10" task-executor="ticketSourceExc" />
</int:service-activator>
<!-- I am getting List<Tickets> -->
<int:splitter input-channel="sourceTicketsSplitter"
output-channel="sourceTicketChannel">
<int:request-handler-advice-chain>
<ref bean="retrier" />
</int:request-handler-advice-chain>
<int:poller fixed-rate="10" />
</int:splitter>
<int:handler-retry-advice id="retrier" max-attempts="4" recovery-channel="myErrorChannel" >
<int:exponential-back-off initial="1000" multiplier="5" maximum="6000"/>
</int:handler-retry-advice>
<int:bridge id="regulatoryBrigde" input-channel="sourceTicketChannel"
output-channel="regulatoryChannel">
<int:poller fixed-rate="10" task-executor="regulatoryExc" />
</int:bridge>
<int:chain id="regulatoryChainFlow" input-channel="regulatoryChannel">
<int:service-activator ref="regulatoryTaskActivator1"
method="process" />
<int:service-activator ref="regulatoryTaskActivator2"
method="process" />
<int:service-activator ref="regulatoryTaskActivator3"
method="process" />
</int:chain>
我已经为拆分器配置重试策略。输出通道是深度为 1 的队列通道（用于测试目的）。我的期望是当我得到一个包含 10 个项目的列表时，它应该在队列 full.It 之后进入异常通道没有发生 way.I 已经将线程睡眠放在 regulatoryTaskActivator1 中以阻止队列
这是正确的，因为发送到输出通道不包括在通知中。这个仅用于 request 部分。请仔细阅读有关问题的文档：https://docs.spring.io/spring-integration/docs/5.0.8.RELEASE/reference/html/messaging-endpoints-chapter.html#message-handler-advice-chain。该建议仅适用于 handleRequestMessage() 方法。
更新
对于您想要处理有限大小的队列并通过重试进行一些错误处理的用例，我建议您将拆分结果输出到 service-activator 周围 @MessagingGateway 在 @Gateway 方法上使用适当的 error-channel 和可能的 @Retryable：
<splitter input-channel="sourceTicketsSplitter"
output-channel="gatewayInputChannel">
<poller fixed-rate="10" />
</splitter>
<service-activator input-channel="gatewayInputChannel" ref="gateway">
<request-handler-advice-chain>
<ref bean="retrier" />
</request-handler-advice-chain>
</service-activator>
<gateway id="gateway" default-request-channel="sourceTicketChannel"/>
如果您的流程是 单向 并且您不希望收到 regulatoryChainFlow 的任何回复，则默认 RequestReplyExchanger 不适合您的要求，您需要使用 void 方法为网关引入一个简单的接口，并将 <gateway> 为此配置为 service-interface.
更新
我忘了告诉你的是，需要使用timeout发送到受限队列。如果没有这样的超时，发件人只需坐下来等待房间，该房间出现在提到的 Thread.sleep(10000);.
那么，你需要的是这样的：
<int:gateway id="stateGateWay" default-request-channel="stateChannel"
default-request-timeout="100"
service-interface="com.biswo.myspringapp.gateway.StateGateWay" />
关注：
<xsd:attribute name="default-request-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
Provides the amount of time dispatcher would wait to send a message.
This timeout would only apply if there is a potential to block in the send call.
For example if this gateway is hooked up to a Queue channel.
Value is specified in milliseconds; it can be a simple long value or a SpEL
expression; array variable #args is available.
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
在此之后我开始在日志中看到 reties 然后 Dispatcher has no subscribers:
2018-10-09 14:53:03.466 TRACE 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : RetryContext retrieved: [RetryContext: count=0, lastException=null, exhausted=false]
2018-10-09 14:53:03.466 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry: count=0
2018-10-09 14:53:03.568 DEBUG 15808 --- [ask-scheduler-3] o.s.r.backoff.ExponentialBackOffPolicy : Sleeping for 100
2018-10-09 14:53:03.669 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Checking for rethrow: count=1
2018-10-09 14:53:03.669 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry: count=1
2018-10-09 14:53:03.771 DEBUG 15808 --- [ask-scheduler-3] o.s.r.backoff.ExponentialBackOffPolicy : Sleeping for 500
2018-10-09 14:53:04.271 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Checking for rethrow: count=2
2018-10-09 14:53:04.271 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry: count=2
2018-10-09 14:53:04.372 DEBUG 15808 --- [ask-scheduler-3] o.s.r.backoff.ExponentialBackOffPolicy : Sleeping for 600
2018-10-09 14:53:04.973 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Checking for rethrow: count=3
2018-10-09 14:53:04.973 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry: count=3
2018-10-09 14:53:05.073 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Checking for rethrow: count=4
2018-10-09 14:53:05.073 DEBUG 15808 --- [ask-scheduler-3] o.s.retry.support.RetryTemplate : Retry failed last attempt: count=4
2018-10-09 14:53:05.077 WARN 15808 --- [ask-scheduler-3] o.s.i.c.MessagePublishingErrorHandler : Error message was not delivered.
org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.errorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.errorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.messaging.MessageDeliveryException: Failed to send message to channel 'stateChannel' within timeout: 100, failedMessage=GenericMessage [payload=State [id=2426, name=Pamanzi, countryId=141], headers={id=710d13df-dfbb-ba95-8d28-4785d1a1e3d3, timestamp=1539111184973}], failedMessage=GenericMessage [payload=State [id=2426, name=Pamanzi, countryId=141], headers={COUNTRY_ID=141, sequenceNumber=2, COUNTRY=com.biswo.myspringapp.model.Country@5fd4ea2d, sequenceSize=2, correlationId=ab423aa4-c80e-6af0-33a8-f57337f25352, id=a70d6b11-fc23-edd9-7973-1342c3fab8a6, timestamp=1539111183466}], headers={id=ee0ffd29-8e6a-ab5a-6ffb-1016fa38457e, timestamp=1539111185074}], failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.messaging.MessageDeliveryException: Failed to send message to channel 'stateChannel' within timeout: 100, failedMessage=GenericMessage [payload=State [id=2426, name=Pamanzi, countryId=141], headers={id=710d13df-dfbb-ba95-8d28-4785d1a1e3d3, timestamp=1539111184973}], failedMessage=GenericMessage [payload=State [id=2426, name=Pamanzi, countryId=141], headers={COUNTRY_ID=141, sequenceNumber=2, COUNTRY=com.biswo.myspringapp.model.Country@5fd4ea2d, sequenceSize=2, correlationId=ab423aa4-c80e-6af0-33a8-f57337f25352, id=a70d6b11-fc23-edd9-7973-1342c3fab8a6, timestamp=1539111183466}], headers={id=ee0ffd29-8e6a-ab5a-6ffb-1016fa38457e, timestamp=1539111185074}], headers={id=34d0e6e6-0af8-1078-5dd7-aa6643ba6c8d, timestamp=1539111185075}] for original GenericMessage [payload=[State [id=2425, name=Mayotte, countryId=141], State [id=2426, name=Pamanzi, countryId=141]], headers={COUNTRY_ID=141, id=ab423aa4-c80e-6af0-33a8-f57337f25352, COUNTRY=com.biswo.myspringapp.model.Country@5fd4ea2d, timestamp=1539111183462}]
仅仅因为您的 errorChannel 是 DirectChannel 并且没有任何消费者。