i use activemq to route message ! the broker is setted as message queue.
the broker received message ,and the web console show "Messages Enqueued<http://113.105.65.86:8161/admin/queues.jsp#> " inceased; and client system consume message , the same time "Messages Dequeued <http://113.105.65.86:8161/admin/queues.jsp#> " number is decreased. today i found that client system consumed message , the same time "Messages Dequeued <http://113.105.65.86:8161/admin/queues.jsp#> " number is decreased; but the client system really can't receive any message! the client system log write as follow: 2014-04-08 08:00:00,044 [ActiveMQ InactivityMonitor ReadCheckTimer] DEBUG [org.apache.activemq.transport.AbstractInactivityMonitor] - 30000ms elapsed since last read check. 2014-04-08 08:00:05,533 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG [org.apache.activemq.transport.AbstractInactivityMonitor] - WriteChecker: 10000ms elapsed since last write check. 2014-04-08 08:00:15,533 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG [org.apache.activemq.transport.AbstractInactivityMonitor] - WriteChecker: 10000ms elapsed since last write check. 2014-04-08 08:00:25,534 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG [org.apache.activemq.transport.AbstractInactivityMonitor] - WriteChecker: 10001ms elapsed since last write check. 2014-04-08 08:00:30,044 [ActiveMQ InactivityMonitor ReadCheckTimer] DEBUG [org.apache.activemq.transport.AbstractInactivityMonitor] - 30000ms elapsed since last read check. 2014-04-08 08:00:35,534 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG [org.apache.activemq.transport.AbstractInactivityMonitor] - WriteChecker: 10000ms elapsed since last write check. 2014-04-08 08:00:45,534 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG [org.apache.activemq.transport.AbstractInactivityMonitor] - WriteChecker: 10000ms elapsed since last write check. 2014-04-08 08:00:55,534 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG [org.apache.activemq.transport.AbstractInactivityMonitor] - WriteChecker: 10000ms elapsed since last write check. when i restart the client system ,the client system received message again,it works again. the client system is long *connection *to the broker ,here is the code: public abstract class ActiveMQMessageConsumer implements MessageListener { private static Logger logger = Logger.getLogger(ActiveMQMessageConsumer.class); protected String destName; private boolean isQueue; private static ConnectionFactory connectionFactory; private Connection connection; private Queue queue; private Destination destination; private MessageConsumer consumer; private Session session; protected MQConfig mqconfig; public void init(MQConfig mqconfig){ this.mqconfig = mqconfig; this.isQueue = true; if(mqconfig !=null){ connectionFactory = new ActiveMQConnectionFactory(mqconfig.getBrokerUser(),mqconfig.getBrokerPwd(),mqconfig.getBrokerURL()); reconnect(); } } public void close(){ try { if(consumer != null){ consumer.close(); } if(session != null){ session.close(); } if (connection != null) { connection.close(); } } catch (JMSException e) { e.printStackTrace(); logger.error(e.getMessage(), e); } } private Connection reconnect(){ close(); if(connectionFactory != null){ try { connection = connectionFactory.createConnection(); connection.start(); session = connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE); queue = new ActiveMQQueue(destName); consumer = session.createConsumer(queue); consumer.setMessageListener(this); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); logger.error(e.getMessage(), e); } return connection; }else { logger.error("create connectionFactory error , reboot connect !"); } return null; } public abstract void work(Message message); @Override public void onMessage(Message message) { if (message != null && message instanceof TextMessage) { work(message); } } }
