JMS - 将侦听器添加到 MessageConsumer 时出错

Question

我正在尝试为 JMS 队列创建 POC。我使用 spring mvc 控制器作为 jms 的客户端。尝试将异步侦听器添加到 MessageConsumer 对象（代码 sn-p）时出现以下错误。我在某处读到只能将侦听器添加到 MDB（消息驱动 bean）中，这是真的吗？

设置：为 JMS 使用 websphere 服务器总线。为 conn factory、destinations 等添加了 jndi，以便同步操作一切正常。但是，对于异步，这是行不通的。

使用this 设置JMS

[1/28/14 14:38:12:570 CST] 0000005d SystemErr     R javax.jms.IllegalStateException: CWSIA0084E: The method MessageConsu
mer.setMessageListener is not permitted in this container.
[1/28/14 14:38:12:572 CST] 0000005d SystemErr     R     at com.ibm.ws.sib.api.jms.impl.JmsMsgConsumerImpl._setMessageListen
er(JmsMsgConsumerImpl.java:660)
[1/28/14 14:38:12:573 CST] 0000005d SystemErr     R     at com.ibm.ws.sib.api.jms.impl.JmsMsgConsumerImpl.setMessageListene
r(JmsMsgConsumerImpl.java:609)

代码：

public void connect(String hostName, String portNumber,

    String connectionFactoryString, String consumerJNDIName)

    throws Exception {

        Hashtable env = new Hashtable();

        env.put(Context.PROVIDER_URL, "iiop://" + hostName + ":" + portNumber
                + "");

        env.put(Context.INITIAL_CONTEXT_FACTORY,
                "com.ibm.websphere.naming.WsnInitialContextFactory");

        InitialContext initialContext = new InitialContext(env);

        ConnectionFactory connectionFactory = (ConnectionFactory) initialContext
                .lookup(connectionFactoryString);

        connection = connectionFactory.createConnection();

        connection.start();

        // create destination - JMSQueue

        Destination destinationReceiver = (Destination) initialContext
                .lookup(consumerJNDIName);

        consumerSession = connection.createSession(false,
                Session.AUTO_ACKNOWLEDGE);

        consumer = consumerSession.createConsumer(destinationReceiver);

        consumer.setMessageListener(new MessageListener() { **// ERROR here**

            public void onMessage(Message msg) {
                try {
                    System.out.println("received: " + ((TextMessage) msg).getText());
                } catch (JMSException ex) {
                    ex.printStackTrace();
                }

            }
        });
    }

Answer 1

要使用来自Queue 的消息，您只需要在Spring context 中注册的MessageListener。

在web.xml 中，您应该在应用服务器中注册对 JMS 资源的资源引用：

<resource-ref>
    <res-ref-name>jms.jndi.cf.name</res-ref-name>
    <res-type>javax.jms.ConnectionFactory</res-type>
    <res-auth>Container</res-auth>
    <res-sharing-scope>Shareable</res-sharing-scope>
</resource-ref>

<resource-ref>
    <res-ref-name>jms.jndi.queue</res-ref-name>
    <res-type>javax.jms.Queue</res-type>
    <res-auth>Container</res-auth>
    <res-sharing-scope>Shareable</res-sharing-scope>
</resource-ref>

下一步是Spring context:

    <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:util="http://www.springframework.org/schema/util" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:jee="http://www.springframework.org/schema/jee"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
   http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd
   http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
   http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.1.xsd
   http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.1.xsd 
   http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee.xsd">

<tx:annotation-driven/>
<context:component-scan base-package="com.jms" />
    <!-- Connection Factory -->
<jee:jndi-lookup id="myConnectionFactory" jndi-name="jms.jndi.cf.name" />

<!-- Queue -->
<jee:jndi-lookup id="destination" jndi-name="jms.jndi.queue.name" />

<!-- JMS Destination Resolver -->
<bean id="jmsDestinationResolver" class="org.springframework.jms.support.destination.DynamicDestinationResolver">
</bean>

<!-- JMS Queue Template -->
<bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory">
        <ref bean="myConnectionFactory" />
    </property>
    <property name="destinationResolver">
        <ref bean="jmsDestinationResolver" />
    </property>
    <property name="pubSubDomain">
        <value>false</value>
    </property>
    <property name="receiveTimeout">
        <value>20000</value>
    </property>
</bean>

<bean id="messageListener" class="com.jms.JMSMessageListener" />

<!-- Message listener container -->
<bean id="jmsConsumerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="myConnectionFactory" />
    <property name="destination" ref="destination" />
    <property name="messageListener" ref="messageListener" />

</bean>
</beans>

最后一步是MessageListener:

package com.jms;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;

import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import java.io.ByteArrayInputStream;
import java.io.StringReader;

public class JMSMessageListener implements MessageListener {

private static final Logger LOGGER = LoggerFactory.getLogger(JMSMessageListener.class);

private static final String ENCODNG = "UTF-8";

@Transactional
public void onMessage(Message message) {
    if (message instanceof BytesMessage) {
        try {
            BytesMessage bytesMessage = (BytesMessage) message;
            LOGGER.debug("Receive message from Queue:\n" + bytesMessage);

            byte[] data = new byte[(int) bytesMessage.getBodyLength()];
            bytesMessage.readBytes(data);
            String stringMessagePayload = new String(data, ENCODNG);
            LOGGER.debug("Message payload: \n" + stringMessagePayload);
            }
        } catch (Exception ex) {
            LOGGER.error("Error has occured: " + ex);
            throw new RuntimeException(ex);
        }
    } else {
        throw new IllegalArgumentException("Message must be of type BytesMessage");
    }
}
}

我希望这个简单的示例对您有所帮助。