Spring 集成:如何在聚合器之后处理服务中的异常

Question

我有一个依赖于Spring Integration(4.0.4.RELEASE(和RabbitMQ的应用程序。我的流程如下：

消息通过进程进入队列(它们不需要任何答案(：网关 -> 通道 -> RabbitMQ

然后被另一个进程排出：

RabbitMQ --1-->入站通道适配器 A --2-->链 B --3-->聚合器 C --4--> 服务激活器 D --5--> 最终服务激活器 E

解释和上下文

具体的事情是，在我的应用程序中，我没有任何地方使用拆分器：聚合器 C 只是等待足够的消息到来，或者超时过期，然后将批处理转发到服务 D。消息可能会卡在聚合器 C 中很长时间，不应被视为在那里使用。只有在服务 D 成功完成后，才应使用它们。因此，我在入站通道适配器 A 上使用手动确认，服务 E 负责确认批处理。

自定义聚合器

我通过重新定义聚合器解决了设置为 AUTO 时的确认问题。实际上，如果流中发生任何异步进程，则会立即确认消息(请参阅此处的问题(。因此，我切换到手动确认并像这样实现聚合器：

     <bean class="org.springframework.integration.config.ConsumerEndpointFactoryBean">
        <property name="inputChannel" ref="channel3"/>
        <property name="handler">
            <bean class="org.springframework.integration.aggregator.AggregatingMessageHandler">
                <constructor-arg name="processor">
                    <bean class="com.test.AMQPAggregator"/>
                </constructor-arg>
                <property name="correlationStrategy">
                    <bean class="com.test.AggregatorDefaultCorrelationStrategy" />
                </property>
                <property name="releaseStrategy">
                    <bean class="com.test.AggregatorMongoReleaseStrategy" />
                </property>
                <property name="messageStore" ref="messageStoreBean"/>
                <property name="expireGroupsUponCompletion" value="true"/>
                <property name="sendPartialResultOnExpiry" value="true"/>
                <property name="outputChannel" ref="channel4"/>
            </bean>
        </property>
    </bean>
    <bean id="messageStoreBean" class="org.springframework.integration.store.SimpleMessageStore"/>
    <bean id="messageStoreReaperBean" class="org.springframework.integration.store.MessageGroupStoreReaper">
        <property name="messageGroupStore" ref="messageStore" />
        <property name="timeout" value="${myapp.timeout}" />
    </bean>
    <task:scheduled-tasks>
        <task:scheduled ref="messageStoreReaperBean" method="run" fixed-rate="2000" />
    </task:scheduled-tasks>

我确实想以不同的方式聚合标头，并保留所有amqp_deliveryTag的最高值，以便以后在服务 E 中进行多连接(请参阅此线程(。到目前为止，这很好用，除了它比典型的聚合器命名空间详细得多(请参阅此旧 Jira 票证(。

服务业

我只是使用基本配置：

链-B

<int:chain input-channel="channel2" output-channel="channel3">
        <int:header-enricher>
            <int:error-channel ref="errorChannel" /> // Probably useless
        </int:header-enricher>
        <int:json-to-object-transformer/>
        <int:transformer    ref="serviceABean" 
                            method="doThis" />
        <int:transformer    ref="serviceBBean" 
                            method="doThat" />
    </int:chain>

服务-D

<int:service-activator  ref="serviceDBean"
                            method="doSomething"
                            input-channel="channel4"
                            output-channel="channel5" />

错误管理

由于我依赖手动确认，因此我还需要手动拒绝消息，以防发生异常。我对入站通道适配器 A 有以下定义：

<int-amqp:inbound-channel-adapter   channel="channel2"
                                            queue-names="si.queue1"
                                            error-channel="errorChannel"
                                            mapped-request-headers="*"
                                            acknowledge-mode="MANUAL"
                                            prefetch-count="${properties.prefetch_count}"
                                            connection-factory="rabbitConnectionFactory"/>

我对错误通道使用以下定义：

<int:chain input-channel="errorChannel">
            <int:transformer ref="errorUnwrapperBean" method="unwrap" />
            <int:service-activator ref="amqpAcknowledgerBean" method="rejectMessage" />
</int:chain>

ErrorUnwrapper 基于此代码，整个异常检测和消息拒绝工作良好，直到消息到达聚合器 C。

问题

如果在处理服务激活器 D 中的消息时引发异常，那么我会看到此异常，但 errorChannel 似乎没有收到任何消息，并且我的 ErrorUnwrapperunwrap(( 方法没有被调用。当抛出异常("ahah"(时，我看到的定制堆栈跟踪如下：

2014-09-23 16:41:18,725 ERROR o.s.i.s.SimpleMessageStore:174: Exception in expiry callback
org.springframework.messaging.MessageHandlingException: java.lang.Exception: ahahaha
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:78)
    at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:71)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:170)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
(...)
Caused by: java.lang.Exception: ahahaha
    at com.myapp.ServiceD.doSomething(ServiceD.java:153)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
(...)
2014-09-23 16:41:18,733 ERROR o.s.s.s.TaskUtils$LoggingErrorHandler:95: Unexpected error occurred in scheduled task.
org.springframework.messaging.MessageHandlingException: java.lang.Exception: ahahaha
(...)

问题

如何告诉处理来自此类聚合器的消息的服务将错误发布到 errorChannel？我试图通过标题丰富器在标头中指定错误通道，但没有运气。我正在使用默认的 errorChannel 定义，但我也尝试更改其名称并重新定义它。我在这里一无所知，即使我找到了这个和那个，我也没有设法让它工作。提前感谢您的帮助！

Answer 1

正如您在 StackTrace 中看到的，您的进程是从 MessageGroupStoreReaper 线程启动的，该线程从默认ThreadPoolTaskScheduler启动。

因此，您必须为此提供自定义 Bean：

<bean id="scheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
    <property name="errorHandler">
        <bean class="org.springframework.integration.channel.MessagePublishingErrorHandler">
            <property name="defaultErrorChannel" ref="errorChannel"/>
        </bean>
    </property>
</bean>
<task:scheduled-tasks scheduler="scheduler">
    <task:scheduled ref="messageStoreReaperBean" method="run" fixed-rate="2000" />
</task:scheduled-tasks>

但是，我看到了在<aggregator>上error-channel的好处，我们确实从不同的分离线程中获得了几个点，我们无法正常处理。