Spring集成:邮件错误和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-amqp:outbound-gateway前面引入一个int-mail:outbound-channel-adapter (请参阅下面的Error Handling配置)，我希望能够调用int:service-activator来构造并返回一个响应给调用者。

我是不是想错了？我看到其他人有一个相似问题，但仍然没有答案。我提到的这两种配置都会发送电子邮件，但总是阻止呼叫者，而不会在超时时收到响应。

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

网关

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

}

服务激活器Config

使用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中间件的复杂性，只是为了最终发送电子邮件。

只有您需要的是以to端点作为订阅者的<publish-subscribe-channel id="defaultErrorChannel">。

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

您可以在Spring 参考手册中找到有关此问题的更多信息。