2

我正在尝试为 JMS 队列编写通道适配器。我想向 JMS 队列发送消息并在 Spring Integration 的频道上接收它。

当我使用普通的 JMS（使用 ActiveMQ）时，一切正常，所以我认为错误存在于我的 Spring 代码中。

所以这是我的 Spring 配置文件：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xmlns:int="http://www.springframework.org/schema/integration"
        xmlns:jms="http://www.springframework.org/schema/integration/jms"
        xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/integration
        http://www.springframework.org/schema/integration/spring-integration.xsd
        http://www.springframework.org/schema/integration/jms
        http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">

    <!-- jms beans -->
    <bean id="jms.msgQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="MSG_QUEUE" />
    </bean>

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

    <!-- spring integration beans -->
    <int:channel id="channels.jms.allMessages">
        <int:queue capacity="1000" />
    </int:channel>

    <jms:inbound-channel-adapter id="adapters.jms.msgAdapter"
        connection-factory="jms.connectionFactory"
        destination="jms.msgQueue"
        channel="channels.jms.allMessages">
        <int:poller fixed-rate="500" />
    </jms:inbound-channel-adapter>

</beans>

这是可以正常工作的普通 JMS 发送-接收代码：

  @Test
    public void testPlainJms() throws JMSException {
        MessageProducer producer = session.createProducer(msgQueue);
        MessageConsumer consumer = session.createConsumer(msgQueue);

        // send to JMS queue
        TextMessage textMessage = session.createTextMessage();
        textMessage.setText("Message from JMS");
        producer.send(textMessage);

        connection.start();
        javax.jms.Message message = consumer.receive(TIMEOUT);
        assertEquals("Message from JMS", ((TextMessage) message).getText());
        connection.stop();
    }

这是使用 Spring 的 MessageChannel 的代码，它通常不起作用（有时会起作用，但我无法确定何时）：

@Test
public void testInboundAdapter() throws JMSException {
    MessageProducer producer = session.createProducer(msgQueue);

    // send to JMS queue
    TextMessage textMessage = session.createTextMessage();
    textMessage.setText("Message from JMS");
    producer.send(textMessage);

    // receive in local channel (using inbound adapter)
    Message<?> received = ((PollableChannel) msgChannel).receive(TIMEOUT);
    String payload = (String) received.getPayload();
    assertEquals("Message from JMS", payload);
}

我收到来自可轮询 msgChannel 的消息时收到NullPointerException 。以下是我如何将 bean 从 Spring 配置自动装配到我的测试类：

@Autowired @Qualifier("jms.msgQueue")
Queue msgQueue;

@Autowired @Qualifier("channels.jms.allMessages")
MessageChannel msgChannel;

@Autowired
ConnectionFactory connectionFactory;
4

2 回答 2

2

奇怪，我能够重现您的问题，可能是入站通道适配器的错误，但我有一些始终有效的东西，通过从inbound-channel-adapterto 更改message-driven-channel-adapter，试试这个：

<jms:message-driven-channel-adapter id="adapters.jms.msgAdapter"
    connection-factory="jms.connectionFactory"
    destination="jms.msgQueue"
    channel="channels.jms.allMessages" />
于 2012-06-24T21:03:57.017 回答
-1

超时时失败。

Message<?> received = ((PollableChannel) msgChannel).receive(TIMEOUT);

你必须received检查null

于 2012-06-24T19:18:18.957 回答