作为共享 Pub/Sub 的 Spring 集成服务接口(interface)网关回复 channel

Question

这类似于 Intermittent BridgeHandler & PublishSubscribeChannel call when gateways' reply channel is pub/sub但情况有所不同，因为回复 channel 不会“丢失”。问题是什么是适合我的场景的最佳解决方案。

我正在使用 Spring 集成来启动 Spring 批处理作业。我有许多输入路线，例如文件轮询和 http 请求。这些都路由到 batch-int 作业启动网关。引用的作业启动器有一个任务执行器，因此作业启动是异步的。此网关在指定 channel 上回复。

<int:gateway service-interface="c.c.c.etl.gateway.JobSubmissionService" id="jobSubmissionService" default-request-channel="jobLauchInputChannel" default-reply-channel="jobLaunchReplyChannel">
</int:gateway>

<int:bridge id="filePollerBridge" input-channel="filePollerOutputChannel" output-channel="jobLauchInputChannel" />

<batch-int:job-launching-gateway request-channel="jobLauchInputChannel" reply-channel="jobLaunchReplyChannel" job-launcher="jobLauncher">
</batch-int:job-launching-gateway>

<int:publish-subscribe-channel id="jobLaunchReplyChannel" />

<int:bridge id="jobLaunchReplyChannelBridge" input-channel="jobLaunchReplyChannel" output-channel="loggingChannel">
</int:bridge>

这个指定的 channel 'jobLaunchReplyChannel' 是 pub/sub 并且有一个记录器在监听它。此 channel 还用作服务接口(interface)网关的回复 channel 。

我遇到的问题是，当通过不是网关的源(例如轮询器)请求作业时，网关添加的桥会抛出异常，因为没有在回复上设置回复 channel 。

我通过向通过网关发送的消息添加 header 并过滤掉仅带有此 header 的消息到新的“gatewayReplyChannel”来解决此问题。

<int:gateway service-interface="c.c.c.etl.gateway.JobSubmissionService" id="jobSubmissionService" default-request-channel="httpJobRequestInputChannel" default-reply-channel="jobSubmissionServiceReplyChannel">
  <int:default-header name="isJobSubmissionServiceMessage" value="true" />
</int:gateway>

<int:channel id="jobSubmissionServiceReplyChannel"></int:channel>

<int:filter id="jobSubmissionServiceReplyChannelFilter" input-channel="jobLaunchReplyChannel" expression="headers.get('isJobSubmissionServiceMessage') == null ? false : headers.get('isJobSubmissionServiceMessage')" output-channel="jobSubmissionServiceReplyChannel"
throw-exception-on-rejection="false" />

有更好的方法吗？

Answer 1

嗯.. 这是一个有趣的问题。主要原因是 MessagingGatewaySupport 为内部 BridgeHandler 创建了 replyMessageCorrelator 端点。

当我们直接向 reply-channel 发送消息时，我们确实有这种奇怪的行为。 BridgeHandler 尝试从 header 向 replyChannel 发送消息。

而且我们真的无法做任何事情来阻止这种逻辑。并且无法保护显式 reply-channel 免受直接消息的影响。

我认为您的解决方案是正确的。克服这个问题的另一种方法:从另一个流程的开始添加 replyChannel header (在您的情况下是文件轮询)，只需使用如下内容:

<header-enricher>
  <reply-channel ref="nullChannel"/>
</header-enricher>

欢迎提出 JIRA就此事提出问题，我们将看看我们能做些什么。至少我们可以记录这些细节。