error-handling - 带有 errorChannel 标头的异步消息流以异常结束
问题描述
我在 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 频道中抛出的,所以应该没问题,但事实证明我错了。我应该非常感谢任何可以解释我的错误的人。
提前致谢!
我试图用任务执行器将 enterArchiveS3FileChannel 变成 pub-sub,但它没有帮助。
解决方案
当您使用网关包装对流的调用时,如果需要,最后一个会在errorChannel
标题旁边填充一个。replyChannel
两者都具有TemporaryReplyChannel
与网关呼叫的自然回复关联的相同实例。
您所有的<int:header-enricher>
s 都没有任何影响,因为已经有一个errorChannel
标题并且默认行为不会覆盖。
要解决此类问题,您只需error-channel
在<gateway>
.
推荐阅读
- visual-studio-code - 突出显示 VSCode 中的自定义代码部分
- amazon-web-services - 带有签名的响应 403 与使用 presignurl s3 不匹配
- firebase - 我可以向 Firebase 中的设备(而不是浏览器)发送消息吗?
- python - 从 Python Discord 机器人中的同步线程发送消息
- javascript - document.elementFromPoint(x, y) 没有报告正确的子元素
- haskell - 了解如何应用 haskell 应用函子
- python - 数据框中的计算和删除/创建列
- javascript - 我如何只允许经过身份验证的用户(使用 cognito)专门访问他们自己的 S3 存储桶/密钥?
- javascript - React Native ScrollView - How to scroll to a child component from another child component button?
- kubernetes - Helm 等待依赖项部署在 kubernetes 上准备就绪