使用WebSphere的Spring集成聚合示例

Question

问题：流输入仅适用于向聚合器发送输出通道输出的1个输入。随后的消息只进入丢弃通道logLateArvers。什么条件被用来发送到丢弃通道？

描述：尝试使用使用WebSphere的聚合器为基本jms移植Spring集成示例。

更新：-打开调试显示轮询器正在工作。消息被拉入并放到MQ，响应被拾取。但是，对于第一组消息之后的MQ场景，不使用AggregatingMessageHandler。消息被发送到丢弃通道与通道“输出”上的“logLateArrivers”适配器进行输出。我将问题声明的措辞改得更具体一些。

Spring集成示例：Spring集成Github示例

使用Spring集成的输出示例：

test1
test2
[TEST1, TEST1]
[TEST2, TEST2]

使用Spring与Websphere集成的输出

test1
test2
[TEST1, TEST2]
[TEST1, TEST2]

使用Websphere MQ移植更改

>

common.xml

<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    <property name="targetConnectionFactory">
        <bean class="com.ibm.mq.jms.MQConnectionFactory">
            <property name="channel" value="channelName" />
            <property name="hostName" value="host1234" />
            <property name="port" value="1111" />
            <property name="queueManager" value="testqmgr" />
            <property name="transportType" value="1" /> 
        </bean>
    </property>
    <property name="sessionCacheSize" value="10"/>
</bean>

<bean id="requestQueue" class="com.ibm.mq.jms.MQQueue">
    <constructor-arg value="requestQueue"/>
</bean>

<bean id="requestTopic" class="com.ibm.mq.jms.MQTopic">
    <constructor-arg value="topic.demo"/>
</bean>

<bean id="replyQueue" class="com.ibm.mq.jms.MQQueue">
    <constructor-arg value="replyQueue"/>
</bean>

<!-- Poller that is the stream in channel for console input -->
<integration:poller id="poller" default="true" fixed-delay="1000"/>

一个ggregation.xml

<int-stream:stdin-channel-adapter id="stdin" channel="stdinToJmsoutChannel"/>

<int:channel id="stdinToJmsoutChannel"/>

<int:chain input-channel="stdinToJmsoutChannel">
    <int:header-enricher>
        <int:header name="jms_replyTo" ref="replyQueue" />
    </int:header-enricher>
    <int-jms:outbound-channel-adapter destination="requestTopic" />
</int:chain>

<int-jms:message-driven-channel-adapter channel="jmsReplyChannel"
    destination="replyQueue"/>

<int:channel id="jmsReplyChannel" />

<int:aggregator input-channel="jmsReplyChannel" output-channel="out"
    group-timeout="5000"
    expire-groups-upon-timeout="false"
    send-partial-result-on-expiry="true"
    discard-channel="logLateArrivers"
    correlation-strategy-expression="headers['jms_correlationId']"
    release-strategy-expression="size() == 2"/>

<int:logging-channel-adapter id="logLateArrivers" />

<!-- Subscribers -->

<int-jms:inbound-gateway request-channel="upcase" request-destination="requestTopic" />

<int-jms:inbound-gateway request-channel="upcase" request-destination="requestTopic" />

<int:transformer input-channel="upcase" expression="payload.toUpperCase()" />

<!-- Profiles -->

<beans profile="default">

    <int-stream:stdout-channel-adapter id="out" append-newline="true"/>

</beans>

<beans profile="testCase">

    <int:bridge input-channel="out" output-channel="queueChannel"/>

    <int:channel id="queueChannel">
        <int:queue />
    </int:channel>

</beans>

Answer 1

消息应该在jms_correlationId上关联。打开DEBUG日志记录并比较示例和您的版本之间的消息流。可能相关id设置不正确。

入站网关使用此逻辑。。。

replyMessage.setJMSCorrelationID(requestMessage.getJMSMessageID());

因此，当发送到聚合器时，与每个请求关联的消息应该获得相同的jms\u correlationId。

您的测试表明这两条消息在某种程度上具有相同的消息id。

编辑

到达的消息具有相同的相关ID（在本例中为头['jms_correlationId']）将被丢弃（晚到达者），除非到期-组-完成后="true"-允许新组开始而不是丢弃。您需要弄清楚为什么第二组具有与第一组相同的相关id。