java - 使用出站适配器配置断路器以处理连接超时问题

Question

<int:service-activator input-channel="toKafka"  ref="conditionalProducerService" method="producerCircuitBreaker">

  <int:request-handler-advice-chain>
       <ref bean="circuitBreakerAdvice1" />
   </int:request-handler-advice-chain>
            </int:service-activator>

  <int:channel id="failedChannel2" />
  <int-kafka:outbound-channel-adapter
                            id="kafkaOutboundChannelAdapter" kafka-producer-context-ref="producerContext" auto-startup="false" channel="toKafka" message-key="kafka_messageKey">
                            <int:poller fixed-delay="1000" error-channel="failedChannel2" />
            </int-kafka:outbound-channel-adapter>


      <int:chain input-channel="failedChannel2">
        <int:transformer expression="'failed:' + payload.failedMessage.payload + ' with ' + payload.cause.message" />
                            <int-stream:stderr-channel-adapter append-newline="true"/>
            </int:chain>

            <bean id="circuitBreakerAdvice1" class="org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice">
                            <property name="threshold" value="2" />                                         
                            <property name="halfOpenAfter" value="12000" />                     
            </bean>

  public Message<?> producerCircuitBreaker(Message<?> payload) {
          throw  new RuntimeException("foo Pro");}

通过以上配置，我们正在尝试：

1.期望将失败的消息传播到错误通道=“ failedChannel2”，这不会发生。因为我无法在控制台中看到转换后的输出。

2.CircuitBreaker正在为ServiceActivator工作（上面针对应用程序相关的异常），但是如何为出站适配器的失败案例配置CB。例如：当连接超时或服务器突然关闭/网络连接问题/某些环境问题，然后才将消息从SI通道发送到外部（kafka）服务器。在这种情况下，我们可以为CB配置出站适配器吗？

根据有关断路器建议的SI文档，请参见下文。

“通常，此建议可能会用于外部服务，否则可能会花费一些时间（例如
作为尝试建立网络连接的超时）”。

请提出实现方法的建议。非常感谢。

更新的配置：
        
    
    

        <int:gateway default-request-channel="toKafka" error-channel="errorChannel"
default-reply-timeout="0" />

 <int:service-activator input-channel="toKafka">
<bean class="com.XXX.ProducerMessageHandler" >
 <constructor-arg ref="producerContext"/>
</bean>
     <int:request-handler-advice-chain>
                                     <ref bean="circuitBreakerAdvice" />
                       </int:request-handler-advice-chain>

<bean id="transformerService1" class="com.XXX.KafkaTransformerTest" />

 <int:transformer input-channel="errorChannel"
                              order="1" ref="transformerService1" method="transformFailed">

                       </int:transformer>  

 public void transformFailed(Message<?> message) {
          APPLOGGER.log("transformer message test" + message);


 public class ProducerMessageHandler extends KafkaProducerMessageHandler{

            public ProducerMessageHandler(KafkaProducerContext kafkaProducerContext) {
                            super(kafkaProducerContext);
                            // TODO Auto-generated constructor stub
            }

            @Override
            public void handleMessageInternal(final Message<?> message) throws Exception {

                            //super.handleMessageInternal(message);
                            throw new RuntimeException("test foo");
            }

日志：

01-05 @ 23：44：18,598调试org.springframework.integration.config.ServiceActivatorFactoryBean $ 1-org.springframework.integration.config.ServiceActivatorFactoryBean$1@6a0ef4b6收到消息：GenericMessage [payload = hello，headers = {timestamp = 1452017658598，id = e0591162-3b93-9bb6-0699-89b15b20e904}]
  调试：-com.XXX.ProducerMessageHandler＃0收到了消息：GenericMessage [payload = hello，标头= {timestamp = 1452017658598，id = e0591162-3b93-9bb6-0699-89b15b20e904}]
异常：org.springframework.messaging.MessageHandlingException：消息处理程序[com.XXX.ProducerMessageHandler＃0]中发生错误；嵌套的异常是java.lang.RuntimeException：测试foo
01-05 @ 23：44：18,606调试org.springframework.integration.channel.PublishSubscribeChannel-在“ toKafka”频道上预先发送，消息：GenericMessage [payload = hello，标头= {timestamp = 1452017658605，id = 61597941-b2f8-314d- 141d-8f2c058dda4d}]
01-05 @ 23：44：18,606调试org.springframework.integration.config.ServiceActivatorFactoryBean $ 1-org.springframework.integration.config.ServiceActivatorFactoryBean$1@6a0ef4b6收到消息：GenericMessage [payload = hello，headers = {timestamp = 1452017658605，id = 61597941-b2f8-314d-141d-8f2c058dda4d}]
  调试：-com.XXX.ProducerMessageHandler＃0接收到消息：GenericMessage [有效载荷=你好，标头= {timestamp = 1452017658605，id = 61597941-b2f8-314d-141d-8f2c058dda4d}]
异常：org.springframework.messaging.MessageHandlingException：消息处理程序[com.XXX.ProducerMessageHandler＃0]中发生错误；嵌套的异常是java.lang.RuntimeException：测试foo
01-05 @ 23：44：18,606调试org.springframework.integration.channel.PublishSubscribeChannel-在频道“ toKafka”上预先发送，消息：GenericMessage [payload = hello，标头= {timestamp = 1452017658606，id = 119afbf1-6104-feb1- eb44-f646aa932277}]
01-05 @ 23：44：18,606调试org.springframework.integration.config.ServiceActivatorFactoryBean $ 1-org.springframework.integration.config.ServiceActivatorFactoryBean$1@6a0ef4b6收到消息：GenericMessage [payload = hello，headers = {timestamp = 1452017658606，id = 119afbf1-6104-feb1-eb44-f646aa932277}]
得到异常：org.springframework.messaging.MessageHandlingException：消息处理程序中发生错误[org.springframework.integration.config.ServiceActivatorFactoryBean$1@6a0ef4b6];嵌套的异常是org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice $ CircuitBreakerOpenException：断路器为org.springframework.integration.config.ServiceActivatorFactoryBean$1@6a0ef4b6打开
01-05 @ 23：44：18,606调试org.springframework.integration.channel.PublishSubscribeChannel-在频道“ toKafka”上预先发送，消息：GenericMessage [payload = hello，标头= {timestamp = 1452017658606，id = 8dafe2e0-8efe-c827- e745-1387e6045e7d}]
01-05 @ 23：44：18,606调试org.springframework.integration.config.ServiceActivatorFactoryBean $ 1-org.springframework.integration.config.ServiceActivatorFactoryBean$1@6a0ef4b6收到消息：GenericMessage [payload = hello，headers = {timestamp = 1452017658606，id = 8dafe2e0-8efe-c827-e745-1387e6045e7d}]
得到异常：org.springframework.messaging.MessageHandlingException：消息处理程序中发生错误[org.springframework.integration.config.ServiceActivatorFactoryBean$1@6a0ef4b6];嵌套的异常是org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice $ CircuitBreakerOpenException：断路器为org.springframework.integration.config.ServiceActivatorFactoryBean$1@6a0ef4b6打开

Answer 1

该建议仅适用于分配给它的端点，而不适用于下游流。不幸的是，kafka模式不允许将其应用于出站通道适配器。我创建了一个JIRA issue for that。

一种解决方法是将KafkaProducerMessageHandler配置为<bean/>，并从ref将其配置为<service-activator/>。然后，您可以应用断路器。

另一个解决方法是使用流入网关。

<int:service-activator ... ref="gw">
    <int:request-handler-advice-chain ...

</int:service-activator>

<int:gateway id="gw" default-request-channel="toKafka" 
         default-reply-timeout="0"
         error-channel="..." ... />

我不确定为什么您不会在错误通道中看到消息；通常，打开DEBUG日志记录将有助于调试这种情况。

编辑

我刚刚对此进行了测试，并且效果很好...

<int:gateway default-request-channel="toKafka" error-channel="errorChannel"
    default-reply-timeout="0" />

<int:service-activator input-channel="toKafka">
    <bean class="com.example.Foo" />
    <int:request-handler-advice-chain>
        <bean class="org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice">
            <property name="threshold" value="2"/>
        </bean>
    </int:request-handler-advice-chain>
</int:service-activator>

编辑2

如果不使用网关，则可以使用队列通道和轮询器来处理它。这对我也很好...

<int:channel id="toKafka">
    <int:queue />
</int:channel>

<int:service-activator input-channel="toKafka">
    <bean class="com.example.Foo" />
    <int:poller error-channel="errorChannel" fixed-delay="1000" />
    <int:request-handler-advice-chain>
        <bean class="org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice">
            <property name="threshold" value="2"/>
            <property name="halfOpenAfter" value="12000"/>
        </bean>
    </int:request-handler-advice-chain>
</int:service-activator>

或者，您可以添加中间流网关。