gpt4 book ai didi

java - Spring Integration - 网关 - 拆分器 - 带有 JMS 的聚合器

转载 作者:行者123 更新时间:2023-12-02 06:57:18 26 4
gpt4 key购买 nike

我正在尝试使用 spring 集成以 JMS 支持的事件驱动方式执行 Gateway --> Splitter-->ServiceActivator --> Aggregator 模式。我希望服务激活器是多线程的,并且任何端点都可以在集群上执行,而不必在原始服务器上执行。我可以在不使用 JMS(使用 SI channel )的情况下在单个 JVM 中实现此功能,但我知道 SI channel 不会帮助我水平扩展,即多个 VM。

这是我到目前为止的配置

    <int:gateway id="transactionGateway" default-reply-channel="transaction-reply"
default-request-channel="transaction-request" default-reply-timeout="10000"
service-interface="com.test.abc.integration.service.ProcessGateway">
</int:gateway>

<int-jms:outbound-gateway id="transactionJMSGateway"
correlation-key="JMSCorrelationID" request-channel="transaction-request"
request-destination="transactionInputQueue" reply-channel="transaction-reply"
reply-destination="transactionOutputQueue" extract-reply-payload="true"
extract-request-payload="true">
<int-jms:reply-listener
max-concurrent-consumers="20" receive-timeout="5000"
max-messages-per-task="1" />
</int-jms:outbound-gateway>

<!-- Inbound Gateway for Splitter -->
<int-jms:inbound-gateway id="splitterGateWay"
request-destination="transactionInputQueue" request-channel="splitter-input"
reply-channel="splitter-output" concurrent-consumers="1"
default-reply-destination="processInputQueue"
max-concurrent-consumers="1" extract-reply-payload="true"
correlation-key="JMSCorrelationID" extract-request-payload="true" />

<!-- Inbound Gateway Invokes Service Activator and Sends response back to
the channel -->
<int-jms:inbound-gateway id="seriveActivatorGateway"
request-destination="processInputQueue" request-channel="process-input"
reply-channel="process-output" concurrent-consumers="1"
default-reply-destination="processOutputQueue"
max-concurrent-consumers="1" extract-reply-payload="true"
correlation-key="JMSCorrelationID" extract-request-payload="true"
max-messages-per-task="1" />

<int-jms:inbound-gateway id="aggregatorGateway"
request-destination="processOutputQueue" request-channel="aggregator-input"
reply-channel="aggregator-output" concurrent-consumers="1"
default-reply-destination="transactionOutputQueue"
max-concurrent-consumers="1" extract-reply-payload="true"
extract-request-payload="true" max-messages-per-task="1"
correlation-key="JMSCorrelationID" />


<int:splitter id="transactionSplitter" input-channel="splitter-input"
ref="processSplitter" output-channel="splitter-output">
</int:splitter>

<int:service-activator id="jbpmServiceActivator"
input-channel="process-input" ref="jbpmService" requires-reply="true"
output-channel="process-output">
</int:service-activator>

<int:aggregator id="transactionAggregator"
input-channel="aggregator-input" method="aggregate" ref="processAggregator"
output-channel="aggregator-output" message-store="processResultMessageStore"
send-partial-result-on-expiry="false">
</int:aggregator>

在使用网关之前,我尝试使用 JMS 支持的 channel ,但该方法也不成功。我现在面临的问题是 Splitter 现在回复到 transactionOutputQueue 。我尝试使用 jms:header-enricher 但没有取得太大成功。我觉得我解决问题/SI 的方法可能存在根本性缺陷。非常感谢任何帮助/指导。

此外,在我上面提供的代码片段中,使用一个简单的内存聚合器,我知道如果我需要在整个集群中实现此功能,我可能需要一个 JDBC 支持的聚合器,但就目前而言,我正在尝试让此模式在单个虚拟机上运行

这是根据 Gary 的评论更新的工作配置

<bean id="processOutputQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="test.com.abc.process.output" />
</bean>

<bean id="transactionOutputQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="test.com.abc.transaction.result" />
</bean>

<bean id="transactionInputQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="test.com.abc.transaction.input" />
</bean>

<int:gateway id="transactionGateway"
default-request-channel="transaction-request" default-reply-timeout="10000"
default-reply-channel="aggregator-output"
service-interface="com.test.abc.integration.service.ProcessGateway">
</int:gateway>

<int:splitter id="transactionSplitter" input-channel="transaction-request"
ref="processSplitter" output-channel="splitter-output">
</int:splitter>


<int-jms:outbound-gateway id="splitterJMSGateway"
correlation-key="JMSCorrelationID" request-channel="splitter-output"
request-destination="processInputQueue" reply-channel="aggregator-input"
reply-destination="processOutputQueue" extract-request-payload="true"
extract-reply-payload="true">
<int-jms:reply-listener
max-concurrent-consumers="20" receive-timeout="5000" />
</int-jms:outbound-gateway>

<!-- Inbound Gateway Invokes Service Activator and Sends response back to
the channel -->
<int-jms:inbound-gateway id="seriveActivatorGateway"
request-destination="processInputQueue" request-channel="process-input"
reply-channel="process-output" default-reply-destination="processOutputQueue"
concurrent-consumers="5" max-concurrent-consumers="10"
extract-reply-payload="true" correlation-key="JMSCorrelationID"
extract-request-payload="true" max-messages-per-task="1" />

<int:service-activator id="jbpmServiceActivator"
input-channel="process-input" ref="jbpmService" requires-reply="true"
output-channel="process-output">
</int:service-activator>


<int:aggregator id="transactionAggregator"
input-channel="aggregator-input" ref="processAggregator"
output-channel="aggregator-output" message-store="processResultMessageStore"
send-partial-result-on-expiry="false">
</int:aggregator>

<bean id="processResultMessageStore"
class="org.springframework.integration.store.SimpleMessageStore" />
<bean id="processResultMessageStoreReaper"
class="org.springframework.integration.store.MessageGroupStoreReaper">
<property name="messageGroupStore" ref="processResultMessageStore" />
<property name="timeout" value="5000" />
</bean>
<task:scheduled-tasks>
<task:scheduled ref="processResultMessageStoreReaper"
method="run" fixed-rate="1000" />
</task:scheduled-tasks>

<int:logging-channel-adapter id="logger"
level="DEBUG" log-full-message="true" />

<int-stream:stdout-channel-adapter
id="stdoutAdapter" channel="logger" />

我将 JMS 管道仅限于 Service Activator,这正是我最初想要的。

基于上述方法,我唯一的问题是,即使我跨多个 VMS 使用聚合器,我是否需要让我的聚合器得到数据库支持(因为它前面的 JMS 网关确保它只接收具有有效相关 ID 的消息?)

问候,

最佳答案

您可能不需要在每个组件之间使用JMS。然而,我们有很多像这样的链接网关的测试用例,并且一切正常。

肯定是某些东西接线不正确。由于您没有显示完整的配置,因此很难推测。

请务必使用最新版本 (2.2.4) 并打开 DEBUG 日志记录并按照流程跟踪消息;只要您的消息有效负载可以跨 JMS 边界识别,就应该很容易找出问题所在。

关于java - Spring Integration - 网关 - 拆分器 - 带有 JMS 的聚合器,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/17157499/

26 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com