Spring Integration：邮件错误和 HTTP 网关响应

Question

我的用例很简单。我想处理由系统无法访问引起的异常，根据配置的重试策略执行重试，在达到重试阈值时发送电子邮件，并将自定义响应返回给调用者。

我面临的挑战是我不能既发送电子邮件又将回复返回给呼叫者。由于我最初使用的是int-mail:outbound-channel-adapter，因此我会期待这种行为，因为这是一个单向组件：

    <int:chain input-channel="defaultErrorChannel"> 
        <int:service-activator id="mailMessageActivator" expression="@mailHandler.process(payload)" />
        <int-mail:outbound-channel-adapter mail-sender="mailSender" />
    </int:chain>   

但是，如果我在int-mail:outbound-channel-adapter 前面引入int-amqp:outbound-gateway（请参阅下面的Error Handling 配置），我希望能够调用int:service-activator 来构造并向调用者返回响应.

我是不是想错了？我看到其他人有一个similar question，但仍未得到答复。我提到的两种配置都会发送电子邮件，但总是阻止调用者在超时时没有收到响应。

以下是我配置的相关部分：

网关

<int:gateway id="customerGateway" service-interface="com.uscs.crm.integration.CustomerGateway" 
        default-request-channel="syncCustomers" default-reply-channel="replySyncCustomers" default-reply-timeout="30000">           
</int:gateway>      

<int:object-to-json-transformer input-channel="syncCustomers" output-channel="outboundRequestChannel" />    

<int-http:outbound-gateway request-channel="outboundRequestChannel" reply-channel="replySyncCustomers"
              url="http://voorhees148.uscold.com:9595/web/customerSync/createCustomer"
              http-method="POST"
              rest-template="restTemplate"
              expected-response-type="com.uscs.crm.model.CustSyncResponseVO" 
              mapped-request-headers="Authorization, HTTP_REQUEST_HEADERS">
    <int-http:request-handler-advice-chain>
        <ref bean="retryWithBackoffAdviceSession" />
    </int-http:request-handler-advice-chain>
</int-http:outbound-gateway>

错误处理

    <int:channel id="defaultErrorChannel"/>
    <int:channel id="errorResponses"/>

    <!-- 
      ExponentialBackOffPolicy.multipler is applied to wait time over each retry attempt
      with a ExponentialBackOffPolicy.maximum configured.        
    -->
    <bean id="retryWithBackoffAdviceSession" class="org.springframework.integration.handler.advice.RequestHandlerRetryAdvice">
        <property name="retryTemplate">
            <bean class="org.springframework.retry.support.RetryTemplate">
                <property name="backOffPolicy">
                    <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
                        <property name="initialInterval" value="2000" />    
                        <property name="multiplier" value="2" />            
                        <property name="maxInterval" value="30000"/>        
                    </bean>
                </property>
                <property name="retryPolicy">
                    <bean class="org.springframework.retry.policy.SimpleRetryPolicy">
                        <property name="maxAttempts" value="3"/>
                    </bean>
                </property>
            </bean>
        </property>
        <property name="recoveryCallback">
            <bean class="org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer">
                <constructor-arg ref="defaultErrorChannel"/>
            </bean>
        </property>
    </bean>

    <bean id="custSyncResponseHandler" class="com.uscs.crm.integration.handler.CustSyncResponseHandler"></bean>
    <int:chain input-channel="defaultErrorChannel" output-channel="replySyncCustomers"> 
        <int:service-activator id="mailMessageActivator" expression="@mailHandler.process(payload)" />
        <int:header-enricher>
            <int:header name="ERROR_ID" expression="T(java.lang.System).currentTimeMillis()"/>
        </int:header-enricher>          
        <int-amqp:outbound-gateway
            exchange-name="error-responses-exchange"
            routing-key-expression="'error.response.'+headers.ERROR_ID"
            amqp-template="amqpTemplate" />
        <!-- Will this service-activator return a response to the caller (int:gateway) using channel `replySyncCustomers`? -->  
        <int:service-activator id="custSyncResponseActivator" expression="@custSyncResponseHandler.process(payload)" />             
    </int:chain>        

    <int-amqp:inbound-gateway queue-names="error-responses" request-channel="errorResponses"  
            connection-factory="rabbitConnectionFactory" acknowledge-mode="AUTO" />

    <int-mail:outbound-channel-adapter channel="errorResponses" mail-sender="mailSender" />

    <!-- (Outbound Channel Adapter/Gateway) rabbit exchanges, queues, and bindings used by this app -->
    <rabbit:topic-exchange name="error-responses-exchange" auto-delete="false" durable="true">
        <rabbit:bindings>
            <rabbit:binding queue="error-responses" pattern="error.response.*"/>
        </rabbit:bindings>
    </rabbit:topic-exchange>
    <rabbit:queue name="error-responses" auto-delete="false" durable="true"/>          

---

解决方案：在@Artem 的帮助下，我得以完成这项工作。以下是我所做的更改。

服务激活器实现（处理ErrorMessage）

关键是返回重构的Message 以及来自ErrorMessage 的所有标头信息的行。

@Override
public Message<CustSyncResponseVO> process(Message<MessagingException> errorMessage) {

    MessagingException errorException = errorMessage.getPayload();

    CustSyncResponseVO custSyncResponse = new CustSyncResponseVO();
    custSyncResponse.setResponseMessage(ExceptionUtils
            .convertToString(errorMessage.getPayload()));

    return MessageBuilder.withPayload(custSyncResponse)
        .copyHeaders(errorMessage.getHeaders())
        .copyHeadersIfAbsent(errorException.getFailedMessage().getHeaders()).build();

}

服务激活器配置

使用 SpEL 引用 #root 上下文来检索 ErrorMessage 而不是默认的 MessagingException (payload) 并将其传递给我在 POJO 上的 process 方法。

    <bean id="custSyncResponseHandler" class="com.uscs.crm.integration.handler.CustSyncResponseHandler" />        
    <int:chain id="errorGatewayResponseChain" input-channel="defaultErrorChannel" output-channel="replySyncCustomers">
        <int:service-activator id="custSyncResponseActivator" expression="@custSyncResponseHandler.process(#root)" />
    </int:chain>

Answer 1

我认为没有理由仅仅为了最终发送电子邮件而在此处引入 AMQP 中间件复杂性。

您只需要<publish-subscribe-channel id="defaultErrorChannel"> 将端点作为它的订阅者。

第一个是单向电子邮件发送<chain>，第二个是custSyncResponseActivator 回复您的<int-http:outbound-gateway>。

您可以在 Spring Integration Reference Manual 中找到更多信息。