0

也许这是一个愚蠢的问题，我错过了一些非常简单的东西，但我真的被卡住了。我正在使用 ActiveMQ 5.5.1 和 SI 2.1.4

我的配置片段：

- -服务器 - -

    <beans:bean id="connectionFactory"   class="org.apache.activemq.ActiveMQConnectionFactory">
    <beans:property name="brokerURL" value="tcp://localhost:61616" />
</beans:bean>


   <beans:bean id="listQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <beans:constructor-arg name="name" value="LIST_QUEUE"/>
    </beans:bean>   

   <beans:bean id="replyListQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <beans:constructor-arg name="name" value="REPLY_LIST_QUEUE"/>
    </beans:bean>             

    <channel id="replyListChannel"/> 
    <channel id="listIn" />
    <channel id="listDriver"/>
    <channel id="listStock"/>    


    <jms:inbound-channel-adapter id="listInJms"
        connection-factory="connectionFactory"
        destination="listQueue"
        channel="listIn"        
        auto-startup="true">
        <poller fixed-rate="3000"/>
    </jms:inbound-channel-adapter>  



    <header-value-router input-channel="listIn" header-name="List"
            default-output-channel="nullChannel">        
        <mapping value="Driver" channel="listDriver" />
        <mapping value="Stock" channel="listStock" />
    </header-value-router>

    <jms:outbound-channel-adapter connection-factory="connectionFactory"
        channel="replyListChannel"
        destination="replyListQueue"
        auto-startup="true">                 
    </jms:outbound-channel-adapter>

- -客户 - -

<beans:bean id="connectionFactory"  class="org.apache.activemq.ActiveMQConnectionFactory">
    <beans:property name="brokerURL" value="tcp://localhost:61616" />
</beans:bean>      

<beans:bean id="requestListQueue" class="org.apache.activemq.command.ActiveMQQueue">
    <beans:constructor-arg value="LIST_QUEUE"/>
</beans:bean>         

<beans:bean id="replyListQueue" class="org.apache.activemq.command.ActiveMQQueue">
    <beans:constructor-arg value="REPLY_LIST_QUEUE"/>
</beans:bean> 


<channel id="requestListChannel">
    <queue capacity="20"/>
</channel>   

<channel id="listStockChannel">
    <queue capacity="20"/>
</channel>   

<channel id="listDriverChannel">
    <queue capacity="20"/>
</channel>       

<channel id="replyListChannel"/>                

<jms:outbound-gateway id="outListGW"
    connection-factory="connectionFactory"
    request-destination="requestListQueue"
    request-channel="requestListChannel"
    reply-destination="replyListQueue"
    reply-channel="replyListChannel" 
    reply-timeout="20000"
    receive-timeout="20000">
    <poller fixed-rate="5000" />
</jms:outbound-gateway>     


<header-value-router input-channel="replyListChannel" header-name="List"
        default-output-channel="nullChannel">        
    <mapping value="Driver" channel="listDriverChannel" />
    <mapping value="Stock" channel="listStockChannel" />
</header-value-router>

然后在代码中的某个地方，我手动执行请求并收听回复通道：

    public static DriverList requestDriverList() {

    Message<String> ldrm = MessageBuilder.withPayload("DriverList request").
            setHeader("List", "Driver").build();            
    try {
        ApplicationContext ctx = 
        new ClassPathXmlApplicationContext("classpath:dmclnt/config/integ-context.xml");
        MessageChannel requestListChannel = 
                ctx.getBean("requestListChannel", MessageChannel.class);
        QueueChannel listDriverChannel = 
                ctx.getBean("listDriverChannel", QueueChannel.class);


        logger.info("Request for DriverList is sent to channel");

        Message dlm = listDriverChannel.receive(20000);

        String xmlDL = (String)dlm.getPayload(); 

        JAXBContext jaxbctx = JAXBContext.newInstance(DriverList.class);
        DriverList dl = (DriverList)jaxbctx.createUnmarshaller().
        unmarshal(newStringReader(xmlDL));
        logger.info("DriverList objct unmarshalled: "+dl.toString());
        return dl;            
    } catch (JAXBException e) {
        logger.error("Error converting xmlDriverList to DriverList object",e);
        return null;
    } catch (RuntimeException e){
        logger.error(e); 
        return null;
    } 
}

但我收到

"MessageTimeoutException: failed to receive JMS response within timeout of: 20000ms"

每时每刻。当我查看服务器日志时，我看到带有正确有效负载的回复已成功从服务器发送到客户端，此外，回复被放入 REPLY_LIST_QUEUE，正如我在 ActiveMQ 管理控制台中看到的那样。没有更多的事情发生！REPLY_LIST_QUEUE With_Correct_Payload中的消息在此队列中处于 Pending 和 Enqueued 状态。没有消息出列。看来，JmsOutboundGateway 没有从回复目标队列中提取消息，尽管接收超时 =“20000 毫秒”延迟远远超过获得回复。我究竟做错了什么？

4

1 回答 1

1

这是因为出站网关对消息相关性有一些期望。我认为，通过您的配置，网关希望服务器在相关 ID 中返回入站消息 ID。它使用消息选择器来接收其回复。

如果您在服务器上使用入站网关，它将为您处理相关性。

您选择在服务器上使用离散通道适配器而不是入站网关是否有任何特殊原因？

您可能需要考虑升级到 2.2.3，其中对出站网关进行了一些改进（但仍需要服务器进行适当的关联）。

编辑：根据您在下面的评论...

如果您确实想使用一对适配器，则在当前配置下，您必须使用 a<header-enricher/>将入站标头复制jms_messageid到jms_correlationId.

或者，在客户端（出站网关）上，将correlation-key属性设置为JmsCorrelationId. 这将导致网关在出站消息上填充该标头，并且您不需要服务器端的任何内容。

于 2013-04-11T12:33:53.677 回答