error-handling - 带有 errorChannel header 的异步消息流以异常结束

Question

我在 xml 中定义了一个流程，如下所示:

<task:executor id="s3ArchivingTaskExecutor" pool-size="10" />

  <task:scheduler id="s3ArchivingScheduler" pool-size="10" />

  <bean id="s3ArchiveResultMessageStore" class="org.springframework.integration.store.SimpleMessageStore" />

  <!-- message store reaper, ultimate timeout = 5 hours -->
  <bean id="s3ArchiveResultMessageStoreReaper" class="org.springframework.integration.store.MessageGroupStoreReaper">
    <property name="messageGroupStore" ref="s3ArchiveResultMessageStore" />
    <property name="timeout" value="18000000" />
  </bean>

  <!-- run reaper every hour -->
  <task:scheduled-tasks scheduler="s3ArchivingScheduler">
    <task:scheduled ref="s3ArchiveResultMessageStoreReaper" method="run" fixed-rate="3600000" />
  </task:scheduled-tasks>

  <int:channel id="enterArchiveS3FileChannel" />

  <!-- copy the sendTo parameter to another header so sendTo can be used within this pipeline -->
  <int:header-enricher input-channel="enterArchiveS3FileChannel"
                       output-channel="loadSuccessfullyArchivedFilesAndSplitMessageChannel">
    <int:header name="subPipelineSendTo" expression="headers['sendTo']" />
    <int:error-channel ref="errorLoggingTerminationChannel" />
  </int:header-enricher>

  <int:channel id="loadSuccessfullyArchivedFilesAndSplitMessageChannel" />

  <int:chain input-channel="loadSuccessfullyArchivedFilesAndSplitMessageChannel"
             output-channel="splittedArchiveS3FilesChannel">
    <int:service-activator ref="loadTransmittedSuccessfullyFileService"
                           method="gettransmittedSuccessfullyFiles" />
    <!-- after this point, no matter success or fail, messages goes back to archiveS3MessageAggregationChannel -->
    <int:header-enricher>
      <int:error-channel ref="archiveS3FilesErrorHandlingChannel" />
    </int:header-enricher>
    <!-- split messages -->
    <int:splitter ref="perSignatureIntegrationFileMessageSplitter"
                  method="splitMessage" />
  </int:chain>

  <!-- parallel processing (fork) -->
  <int:publish-subscribe-channel id="splittedArchiveS3FilesChannel"
                                 task-executor="s3ArchivingTaskExecutor" />

  <int:chain input-channel="splittedArchiveS3FilesChannel"
             output-channel="archiveS3MessageAggregationChannel">
    <int:service-activator ref="archiveSignatureIntegrationFileService"
                           method="archiveFileForPayload" />
    <int:service-activator ref="updateFileToArchivedByS3KeyService"
                           method="updateFileToArchivedByS3Key" />
    <int:header-enricher>
      <int:error-channel ref="s3ArchiveJobFailedUpdateJobStatusChannel" />
    </int:header-enricher>
  </int:chain>

  <!-- error handling - copy the corelation id and sequence size -->
  <int:channel id="archiveS3FilesErrorHandlingChannel" />

  <int:header-enricher input-channel="archiveS3FilesErrorHandlingChannel"
                       output-channel="archiveS3MessageAggregationChannel">
    <int:header name="correlationId" expression="payload.failedMessage.headers['correlationId']" />
    <int:header name="sequenceSize" expression="payload.failedMessage.headers['sequenceSize']" />
  </int:header-enricher>

  <!-- join back with aggregator -->
  <int:channel id="archiveS3MessageAggregationChannel" />

  <!-- group timeout is sequence size (total amount of messages) * 5 minutes -->
  <int:aggregator id="s3ArchiveMessageAggregator"
                  input-channel="archiveS3MessageAggregationChannel"
                  output-channel="s3ArchiveMessageAggregatorOutputChannel"
                  message-store="s3ArchiveResultMessageStore"
                  ref="archiveS3FilesMessageAggregator"
                  method="aggregate" />

  <int:channel id="s3ArchiveMessageAggregatorOutputChannel" />

  <!-- restore sendTo header -->
  <int:header-enricher input-channel="s3ArchiveMessageAggregatorOutputChannel"
                       output-channel="sendToHeaderValueRouterChannel">
    <int:header name="sendTo" expression="headers['subPipelineSendTo']" />
  </int:header-enricher>

在单元测试中，我用网关包装了这个流程，并为archiveSignatureIntegrationFileService 提供了一个模拟，它总是抛出RuntimeException。我期待 archiveS3FilesErrorHandlingChannel 接收消息，但单元测试失败了

java.lang.RuntimeException: test fail

    at <PROJECT CLASS PATH>.ArchiveSignatureIntegrationFileServiceImpl.archiveFileForPayload(ArchiveSignatureIntegrationFileServiceImpl.java:60)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:171)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:120)
    at org.springframework.integration.handler.support.MessagingMethodInvokerHelper$HandlerMethod.invoke(MessagingMethodInvokerHelper.java:1115)
    at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.invokeHandlerMethod(MessagingMethodInvokerHelper.java:624)
    at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:491)
    at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:362)
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:106)
    at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:93)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:123)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169)
    at org.springframework.integration.handler.MessageHandlerChain.handleMessageInternal(MessageHandlerChain.java:110)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:162)
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:224)
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.access$000(BroadcastingDispatcher.java:56)
    at org.springframework.integration.dispatcher.BroadcastingDispatcher$1.run(BroadcastingDispatcher.java:204)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)

我认为异常是从带有任务执行器的 pub-sub channel 中抛出的，所以应该没问题，但结果证明我错了。我应该非常感谢任何可以解释我的错误的人。

提前致谢!

我试图用任务执行器将 enterArchiveS3FileChannel 变成 pub-sub，但它没有帮助。

Answer 1

当您使用网关包装对流的调用时，最后一个会填充 errorChannel标题旁边有 replyChannel如果需要的话。两者都具有相同的TemporaryReplyChannel例如与网关调用的自然回复关联。

你所有的<int:header-enricher> s 没有任何影响，因为已经有 errorChannel header 和默认行为不会覆盖。

要解决此类问题，您只需配置 error-channel在您的 <gateway> 上.