【发布时间】:2016-07-15 02:50:15
【问题描述】:
我有一个涉及异步执行、从网关返回值到控制器、返回值后继续集成流的 Spring 集成流程。
这里是网关:
@MessagingGateway
public interface GW {
@Gateway(requestChannel = "f.input")
Task input(Collection<MessengerIncomingRequest> messages);
}
这是流程:
@Bean
IntegrationFlow jFlow() {
return IntegrationFlows.from(
MessageChannels.executor("f.input", executor()))
.split()
.channel(MessageChannels.executor(executor()))
.transform(transformer)
.channel(routerChannel())
.get();
}
@Bean
ThreadPoolTaskExecutor executor() {
ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
...
return pool;
}
@Bean
MessageChannel routerChannel() {
return MessageChannels
.publishSubscribe("routerChannel", executor())
.get();
}
@Bean
IntegrationFlow routerChannelFlow() {
return IntegrationFlows
.from(routerChannel())
.publishSubscribeChannel(s -> s
.subscribe(f -> f.bridge(null))
.subscribe(process()))
.get();
}
@Bean
IntegrationFlow process() {
return f ->
f.route(p -> p.getKind().name(),
m -> m.suffix("Channel")
.channelMapping(TaskKind.CREATE.name(), "create")
....
}
@Bean
IntegrationFlow createFlow() {
return IntegrationFlows.from(
MessageChannels.direct("createChannel"))
.handle(routerService)
.get();
}
如何为整个流程定义错误处理程序?最佳实践是什么?我知道我可以为网关方法调用放置一个 try/catch 块,但它只会捕获 jFlow 流中出现的异常,用于 channel(routerChannel()) 之前的所有内容。
如何处理流程其余部分的错误?还是整个流程?
更新
我为publishSubscribeChannel 添加了错误处理程序
@Bean
IntegrationFlow routerChannelFlow() {
return IntegrationFlows
.from(routerChannel())
.publishSubscribeChannel(s -> s
.subscribe(f -> f.bridge(null))
.subscribe(process())
.errorHandler(errorHandler))
.get();
}
但这似乎没有帮助,因为如果出现异常,我会收到以下错误:
cMessagingTemplate$TemporaryReplyChannel : Reply message received but the receiving thread has already received a reply:ErrorMessage [payload=org.springframework.messaging.MessageHandlingException:
我的错误处理程序没有被调用。
更新
根据 Gary 的回答,我尝试了以下代码:
@Bean
IntegrationFlow jFLow() {
return IntegrationFlows.from(
MessageChannels.executor("f.input", executor()))
.split()
.channel(MessageChannels.executor(executor()))
.transform(transformer)
.channel(routerChannel())
.get();
}
@Bean
IntegrationFlow exceptionOrErrorFlow() {
return IntegrationFlows.from(
MessageChannels.direct("exceptionChannel"))
.handle(errorHandler, "handleError")
.get();
}
@Bean
MessageChannel exceptionChannel() {
return MessageChannels.direct("exceptionChannel")
.get();
}
@Bean
IntegrationFlow process() {
return f ->
f.enrichHeaders((spec) ->
spec.header("errorChannel", "exceptionChannel", true))
f.route(p -> p.getKind().name(),
m -> m.suffix("Channel")
.channelMapping(TaskKind.CREATE.name(), "create")
....
}
@MessagingGateway(errorChannel = "exceptionChannel")
在另一次编辑后,我将exceptionChannel 添加到网关,并将丰富的标题移动到我的流程的第二段(异步)。如果在流程的同步部分抛出异常,控制器仍然会被阻塞。
【问题讨论】:
标签: java spring spring-integration