ActiveMQ的消息接收的问题(使用MessageListener)

ActiveMQ

解决方案 »

免费领取超大流量手机卡，每月29元包185G流量+100分钟通话, 中国电信官方发货

一般都是在客户端关闭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) {
      String errorMessage = "JMSException while queueing HTTP JMS Message";
      throw new EventRuntimeException(errorMessage, e);
    } finally {
      SafeCloseUtil.close(producer); // 这里关闭 producer
      SafeCloseUtil.close(session);  // 这里关闭 producer
      SafeCloseUtil.close(connect);
    }
}
服务端只需要实现MessageListener的onMessage,客户端的 session服务端根本不知道，也无法关闭。
@Override
public void onMessage(Message message) {
    printLogMessage("start public function onMessage()..");
    try {
      if (message instanceof ObjectMessage) {
        ObjectMessage objMsg = (ObjectMessage) message;
        Event event = (Event) objMsg.getObject();
        // if Re deliver message warn the message
        if (message.getJMSRedelivered()) {
          log.warn("...", event.getClass().getSimpleName());
        }
        // out Put Event Log
        outPutEventLog(event);
        // dispatch Event
        dispatchEvent(event);
      } else {
        log.error("This MDB message was not instance of ObjectMessage; ignoring.");
      }      printLogMessage("end public function onMessage()..");
    } catch (JMSException e) {
      String errorMessage = "JMS Exception while Listener message.errorMessage:"
          + e.getMessage();
      log.error(errorMessage);
      throw new EventRuntimeException(errorMessage, e);
    }
}
感谢1楼的回复和2楼的...捧场，呵呵不过，我还是有些不太明白。
首先，你说的服务端和客户端，
因为你说服务端实现onMessage接口，
所以我是不是可以理解为
你说的服务端算是消息的接收端，
而客户端是消息的发送端那么，
也就是说，
一旦消息接收端的程序开始运行，开始接收消息，
就始终要开着这个链接，准备消息发送端随时发来的消息？但是，我这个工程里，
消息接收端是一个任务(task）对象，
一段时间内只处理一部分发送过来的消息。
(接收固定数量的消息的方法我在ActiveMQ自带的例子里看到过，方法也了解)
按照上面我"一旦"的逻辑，
是不是这个任务在第一次执行的时候就一直执行下去了？
而不会真正的推出，
而受限于jdk的timer的运行机制，(第一个task运行完了，才安排第2个task运行)
这个任务(task)就是不是始终不会被第二次执行了？比如：
这个任务在第一次被执行的时候，
取得一个consumer->设定了onMessage->被触发后，
这个任务就一直等待发送端发送过来的消息，
而因为它始终是处于活态(忘了什么词了，反正就是一直存在的意思)
timer就不会安排第2次，第3次启动？那么我这个task就没有意义了。
我是不是要考虑其他的方式了呢？
或者直接干脆就执行这个任务一次(直接执行run方法，代码也不改了)？？...
还是说当发送端关闭了这个session，
接收端的各种资源也就释放了？
(应该不会这样吧？
因为发送端和接收端都隔着一个ActiveMQ，
难道ActiveMQ会通知他的其他弟兄关闭关于自己的接收？)一大嘟噜问题，真是不好意思
有空给在下点拨点拨呗
谢谢了good luck
请问对于activemq接受消息的方法：Receive（）和Listener，有什么区别啊？是不是读取了一条消息就会关闭，还是能读取所有的匹配消息啊？对于Topic和Queue两种类型，读取消息，有什么不同啊？我就是创建了一个Topic的持久的订阅者，打开以后，程序到Receive（）方法，那就一直停在那里。。Listener方法也一样，他们为什么不能收到离线是的消息呢？只能收到在线消息
con，和session都可以交给spring去管，con只有1个，session用对象池
<bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
destroy-method="stop">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL"
value="failover:(tcp://10.60.60.107:61616?wireFormat.maxInactivityDuration=0)&maxReconnectDelay=1000" />
</bean>
</property>

<property name="maxConnections" value="1" />

<property name="maximumActive" value="500" />
</bean>
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />

<property name="explicitQosEnabled" value="true" />

<property name="deliveryMode" value="1" />
</bean>