检查ActiveMQ代理的状态

Question

我想创建一个java类,其唯一目的是检查ActiveMQ代理的状态(或连接到ActiveMQ代理,因为中断可能被定义为客户端也失去网络连接).

所以基本上会有一个线程在每隔几秒钟后运行以检查代理的状态,如果有经纪人关闭,我想做一些特定的邮件支持组和类似的任务.

在线示例不够详细,无法解释如何实现上述目标.

有人已经做过这个,或者可以建议一个很好的方法来实现这一点？

谢谢,Neeraj

Answer 1

以下内容还可用于检查ActiveMQ是否已启动并正在运行:

try {
   ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
   // set transport listener so that active MQ start is notified.
   factory.setTransportListener(transportListenerObject); 
   Connection connection = factory.createConnection();
   // This will throw error if activeMQ is not running.
   connection.setClientID("my_client_id"); 
} catch (JMSException ex) {
    if (ex.getLinkedException() instanceof IOException) {
        // ActiveMQ is not running. Do some logic here.
        // use the TransportListener to restart the activeMQ connection
        // when activeMQ comes back up.
    } else {
        // Something seriously went wrong with the factory or connection
        // creation. Abort the process here, as nothing can be done.
        // Log the error and troubleshoot.
    }
}

Answer 2

向代理发送Testmessage:

try {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    Connection conn = factory.createConnection(user, password);
    Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    MessageProducer producer = session.createProducer("test");
    MessageConsumer consumer = session.createConsumer("test");
    consumer.setMessageListener(this); // class that implements MessageListener
    conn.start();
    TextMessage message = new ActiveMQTextMessage();
    message.setText("TestMessage");
    producer.send(message);
} catch (JMSException e) {
    // somethings very wrong
}

连接,发送消息,如果你收到消息:一切都很好.如果不....

我就是做这个的.另外我做了一些其他的事情:

• 听取几个咨询主题,以接收重要事件(如Advisory.FULL),这些事件是一些重要指标,有些事情是错误的.
• 定期从统计插件中获取代理统计信息,以监视消息内存大小和消息存储.
• 配置一个死信队列,以便我知道消息被消息拒绝的时间.

Answer 3

我需要针对同一个问题的解决方案,这就是为什么我会阅读更多关于它并进行多次测试的原因.

在某些环境中发送测试消息(如Laures建议的那样)可能是个问题.

"正常"方式是设置TransportListener(如Anand建议的那样),但实际上实现了提供的接口并对报告的事件做出反应.

对于其他ActiveMQ新手(就像我上个月一样),我发布了一个示例启动实现.它只是为每个事件写入日志记录.在真实的环境中,人们可以考虑transportInterupted()直到transportResumed()或类似的重新连接试验以及更多的事情......

import java.io.IOException;

import org.apache.activemq.transport.TransportListener;
import org.apache.log4j.Logger;

class ConnectionStateMonitor
  implements TransportListener
{
  private static final Logger log = Logger.getLogger(ConnectionStateMonitor.class);

  @Override
  public void onCommand(Object command)
  {
    log.debug("Command detected: '" + command + "'");
  }

  @Override
  public void onException(IOException exception)
  {
    log.error("Exception detected: '" + exception + "'");
  }

  @Override
  public void transportInterupted()
  {
    log.error("Transport interuption detected.");
  }

  @Override
  public void transportResumed()
  {
    log.info("Transport resumption detected.");
  }
}

TransportListener可以设置为:

ActiveMQConnection connection = (ActiveMQConnection) _factory.createConnection();
...
connection.addTransportListener(new ConnectionStateMonitor());

玩得开心!