0

我们正处于应用程序的设计阶段。我们决定使用 Spring-Integration。作为起点，应用程序使用入站适配器从 JMS 队列中读取消息，入站适配器基于轮询和使用任务执行器的多线程。在对消息进行审计记录后，这些接收线程将消息放到通道上，工作线程从通道中拾取每条消息以进行进一步处理。进一步处理本身包括不同的组件，如消息解析、节点对象构建、预链接和链接。在步骤中，它包括用于消息保存和节点对象保存的接收器。这是根据以下配置。

<int-jms:inbound-channel-adapter jms-template="jmsTemplate"  channel="channel1" id="MessageReceiver">
    <int:poller fixed-delay="100" time-unit="MILLISECONDS" task-executor="taskExecutor"/>
</int-jms:inbound-channel-adapter>
<int:channel id="channel1">
        <int:interceptors>
            <int:wire-tap channel="channel.tbl_message"/>
        </int:interceptors>
    </int:channel>

<bean id="dataSource" destroy-method="close"
    class="org.apache.commons.dbcp.BasicDataSource">
    <property name="driverClassName" value="${db.driverClassName}" />
    <property name="url" value="${db.host.url}" />
    <property name="username" value="${db.username}" />
    <property name="password" value="${db.password}" />
</bean>

<int-jdbc:outbound-channel-adapter channel="channel.tbl_message" data-source="dataSource"
    query="#{message.receiver.insert.query}" id="MessageLogger"></int-jdbc:outbound-channel-adapter>

<int:transformer input-channel="channel1" output-channel="channel2" method="parse" id="NodeBuilder">
    <bean class="com.recon.parser.NodeBuilder"></bean>
</int:transformer>
<int:channel id="channel2"/>

<int:filter input-channel="channel3" output-channel="channel5"
    discard-channel="channel4" method="validate" id="NodeValidator">
    <bean class="com.recon.util.Validator"></bean>
</int:filter>
<int:channel id="channel3">
    <int:interceptors>
        <int:wire-tap channel="channel.tbl_node"/>
    </int:interceptors>
</int:channel>
<int-jdbc:outbound-channel-adapter channel="channel.tbl_node" data-source="dataSource"
    query="#{valid.node.insert.query}" id="ValidNodePersist"></int-jdbc:outbound-channel-adapter>

<int:channel id="channel4"/>
<int-jdbc:outbound-channel-adapter channel="channel4" data-source="dataSource"
    query="#{validation.failure.insert.query}" id="FailedNodePersist"></int-jdbc:outbound-channel-adapter>

<int:transformer input-channel="channel2" output-channel="channel3" method="nodeEnricher" id="NodeEnricher">
    <bean class="com.recon.processor.NodeEnricher"></bean>
</int:transformer>

<int:channel id="channel5"/>
<int:service-activator input-channel="channel5" id="LinkerManager">
    <bean class="com.recon.manager.LinkerManager"></bean>
</int:service-activator>

现在我有两个选择：1）创建一个工作线程的线程池。每个工作线程将从节点处理器开始处理消息，节点处理器将以简单的 spring 依赖注入方式使用所有后续组件，而不使用 spring-integration。2）使用task-executor创建每个组件的线程池。每个组件的每个线程都会选择缓冲到其先前通道的输入对象。但是，这将创建许多线程，因为每个组件都将是多线程的。

任何人都可以就选择的方法提出一些解决方案吗？

4

2 回答 2

0

一旦您从 开始您的流程JMS message poller，我认为没有理由将消息从轮询线程转移到任何其他线程。如果您的一般逻辑将在该单个线程中完成会更好，在这种情况下，您的应用程序的吞吐量将取决于 Poller TaskExecutor 并发性。

audit logging，message saving并且node object saving可能从不同的广告中完成。在这种情况下，这QueueChannel是给你的。我的意思是所有这些<int:wire-tap>渠道。

于 2014-05-26T06:57:06.737 回答
0

@prasadsh 您是否通过将<wire-tap>频道转换为 QueueChannel 来获得所需的性能？您也可以尝试将 channel1 转换为 QueueChannel，因为无论如何您都有 channel.tbl_message 来记录来自 jms 的输入消息。

我很想知道您是否有任何解决方案有效，因为我们有类似的要求（10k 消息/秒）

于 2014-05-29T02:19:28.290 回答