日期:2014-05-20  浏览次数:20754 次

ActiveMQ的消息接收的问题(使用MessageListener)
在用ActiveMQ的MessageListener消息接收模式的时候,
我setMessageListener了本类,也就是consumer.setMessageListener(this);
然后,又在onMessage里面做了相应消息的处理。
然后提交了session。

我的问题是,
只关闭这个session,
这个class会结束并进入回收环节吗?

因为我这个类是一个TimerTask,
每隔一段时间启动一次来接收消息,
我想知道的是,如果第一次收到消息,调用了onMessage方法以后,
这个任务会不会就彻底结束了?

还是
不只要在onMessage方法里关闭session对象,
还要在里面关闭Connection对象,
这个任务才算完全的执行完。


如果不关,这个链接会不会一直存在,直到整个程序停止,或者ActiveMQ停止了,
才能结束?
这样的话,我接受的内容如果一多,那还不把内存撑暴啊?

PS:
打字的时候突然想到,
这个conn如果设置成static的,
取得conn也在static块里面,
是不是再执行这个任务的时候,
就不用每次都建立链接,
而直接从这个链接里取得session,然后再取consumer就可以了呢?
这个相当于长链接的conn会不会成为系统里的结石??

贴些代码
就是红色的XXXX这里要不要关掉那个conn对象。

import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * 
 * @author
 * 
 *  1. 数据接收线程监听ActiveMQ队列,并读取消息。。
 * 
 */
public class ReceiveTask extends TimerTask implements MessageListener {

private static final Log log = LogFactory.getLog(ReceiveTask.class);

private Session session;

private javax.jms.Connection conn = null ;

private MessageConsumer consumer = null;

/**
 * 
 */
public void run() {

//开始监听设置
try {

log.info("数据消费者开始启动...");

conn = MQConnection.getJmsConnection();

conn.start();

// 得到session
session = conn.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

// 通过本地节点的接收队列名,监听需要监听的队列名
Destination destination = session.createQueue("QueueName");

MessageProducer replyProducer = session.createProducer(null);

// 设置传送模式为“非持久性消息”
replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

// 初始化消息的消费者
consumer = session.createConsumer(destination);

// 设置监听器,有消息来得时候回调MessageListener.onMessage方法
consumer.setMessageListener(this);

log.info("数据消费者启动完成...");

} catch (Exception e) {
log.info("数据消费者启动异常...");
}
}

/**
 * MessageListener.onMessage
 */
public void onMessage(Message message) {

try {
// 各种处理,略

//提交session
session.commit();
}
} catch (Exception e) {

MQConnection.rollback(session);

MQUtil.log(e, log);
} finally {
// XXXX
}

}
}
ActiveMQ

------解决方案--------------------
一般都是在客户端关闭seesion conection,代码例子

@Override
public void convertAndSend(Event message) {
    // build ConnectionFactory And Queue is necessary
    buildConnectionFactoryAndQueue();
    
    Connection connect = null;
    Session session = null;
    MessageProducer producer = null;
    try {
      connect = jmsConnectionFactory.createConnection();
      session = connect.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
      producer = session.createProducer(queue);
      // create a JMS message and send it
      ObjectMessage objMsg = session.createObjectMessage(message);
      // set message selector
      String messageSelector = message.getMessageSelector();
      objMsg.setStringProperty("messageReceiver", messageSelector);

      producer.send(objMsg);
    } catch (JMSException e) {