我正在尝试构建一个简单的流程,我的流程从通过 HTTP 入站通道适配器接收 HTTP 发布请求并将其发布到“SubscribableChannel”开始。订阅此频道的消费者可能有“N”个。下图显示了流程。
我正在尝试使用 Spring DSL 来配置此流程,但在使其正常工作时遇到了麻烦。下面是我的代码。
@Bean
public IntegrationFlow receiveHttpPost() {
return IntegrationFlows.from(Http.inboundChannelAdapter("/receive")
.mappedRequestHeaders("*")
.requestChannel(httpInAdapterPubSubChannel()))
.transform(new ObjectToStringTransformer())
.get();
}
@Bean
public SubscribableChannel httpInAdapterPubSubChannel()
{
return MessageChannels.publishSubscribe("httpInAdapterPubSubChannel")
.get();
}
@Bean
public IntegrationFlow subscriber1() {
return IntegrationFlows.from(httpInAdapterPubSubChannel())
.handle( message -> System.out.println("Enrich Headers based on Payload...."))
.get();
}
@Bean
public IntegrationFlow subscriber2() {
return IntegrationFlows.from(httpInAdapterPubSubChannel())
.handle( message -> System.out.println("Save Payload to Audit Table..."))
.get();
}
当我运行这个流程时,我得到"Failed to handle Message; nested exception is org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available"。
o.s.i.channel.PublishSubscribeChannel : preSend on channel 'httpInAdapterPubSubChannel', message: GenericMessage [payload=Test, headers={content-length=4, http_requestMethod=POST, accept-language=en-US,en;q=0.8, accept=*/*, host=localhost:8080, http_requestUrl=http://localhost:8080/receive, connection=keep-alive, content-type=text/plain;charset=UTF-8, id=2c6ee729-96ee-1ae5-be31-a9bc56092758, cache-control=no-cache, accept-encoding=gzip, deflate, br, user-agent=Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36, timestamp=1484457726393}]
o.s.i.t.MessageTransformingHandler : org.springframework.integration.transformer.MessageTransformingHandler#0 received message: GenericMessage [payload=Test, headers={content-length=4, http_requestMethod=POST, accept-language=en-US,en;q=0.8, accept=*/*, host=localhost:8080, http_requestUrl=http://localhost:8080/receive, connection=keep-alive, content-type=text/plain;charset=UTF-8, id=2c6ee729-96ee-1ae5-be31-a9bc56092758, cache-control=no-cache, accept-encoding=gzip, deflate, br, user-agent=Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36, timestamp=1484457726393}]
o.a.c.c.C.[.[.[/].[dispatcherServlet] : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.messaging.MessagingException: Failed to handle Message; nested exception is org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available] with root cause
org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:287) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:212) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:129) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:236) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.integration.dispatcher.BroadcastingDispatcher.dispatch(BroadcastingDispatcher.java:185) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.5.RELEASE.jar:4.3.5.RELEASE]
很明显,我在这里做错了什么。我试图找到显示“发布订阅频道”VIA Spring Integration DSL OR Java Configuration的示例。不幸的是,我找不到任何:-/。如果有人可以为我提供一个示例并帮助我找出我的流程有什么问题,我将不胜感激。
我所做的另一个观察是,“当我删除订阅者 'subscriber1' 和 'subscriber2'”时,我仍然遇到同样的错误。这意味着,我在配置 y HttpInboundAdapter 时做错了什么。
此外,如果我将“httpInAdapterPubSubChannel”切换为直接,并且只有一个路由流(没有分支),一切正常。