ActiveMQ 通过JMX监控Connection，Queue，Topic的信息

longgangbai

浏览: 7250344 次
性别:
来自: 上海

最近访客更多访客>>

paladin1988

ljq867

yuxiangtong

chen88358323

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

ActiveMQ学习和研究

ActiveMQ Apache JMS Bean 配置管理

How can I monitor ActiveMQ

In ActiveMQ 4.x you can monitor the broker to see what destinations are being used, their activity along with connections and subscriptions using the following tools

JMX and a JMX console such as jConsole
The Web Console
the Advisory Message feature (using JMS messages to monitor the system)
The Command Agent; ActiveMQ.Agent topic that you query for status
The Visualisation plug-in
The Statistics plug-in (from 5.3)

package easyway.app.activemq.demo.monitors;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.management.ObjectName;
import javax.management.QueryExp;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.jmx.BrokerView;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.ConnectionViewMBean;
import org.apache.activemq.broker.jmx.ConnectorViewMBean;
import org.apache.activemq.broker.jmx.DestinationViewMBean;
import org.apache.activemq.broker.jmx.DurableSubscriptionViewMBean;
import org.apache.activemq.broker.jmx.JobSchedulerViewMBean;
import org.apache.activemq.broker.jmx.ManagedRegionBroker;
import org.apache.activemq.broker.jmx.ManagementContext;
import org.apache.activemq.broker.jmx.NetworkBridgeViewMBean;
import org.apache.activemq.broker.jmx.NetworkConnectorViewMBean;
import org.apache.activemq.broker.jmx.QueueViewMBean;
import org.apache.activemq.broker.jmx.SubscriptionViewMBean;
import org.apache.activemq.broker.jmx.TopicViewMBean;
import org.apache.activemq.broker.region.Queue;
import org.apache.activemq.broker.region.policy.ConstantPendingMessageLimitStrategy;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.usage.MemoryUsage;
import org.apache.activemq.usage.StoreUsage;
import org.apache.activemq.usage.SystemUsage;
import org.apache.activemq.usage.TempUsage;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 监控ActiveMQ的各种信息Broker,Connection,Queue,Topic的数量和压栈和出栈
 * @author longgangbai
 *
 */
public class ActiveMQMonitor {
	   private static final transient Logger LOG = LoggerFactory.getLogger(DestinationSourceMonitor.class);
		
	    protected static final int MESSAGE_COUNT = 2000;
	    protected BrokerService brokerService;
	    protected Connection connection;
	    protected String bindAddress ="tcp://localhost:61619";
	    //ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL;
	    protected int topicCount;

        /**
         * 获取Broker 的AdminView对象
         * @return
         * @throws Exception
         */
    	public BrokerViewMBean getBrokerAdmin() throws Exception {
    		return brokerService.getAdminView();
    	}
    	/**
    	 * 获取所有的QueueViewMBean的
    	 * @return
    	 * @throws Exception
    	 */
    	public Collection<QueueViewMBean> getQueues() throws Exception {
    		BrokerViewMBean broker = getBrokerAdmin();
    		if (broker == null) {
    			return Collections.EMPTY_LIST;
    		}
    		ObjectName[] queues = broker.getQueues();
    		return getManagedObjects(queues, QueueViewMBean.class);
    	}
    	
    	
        /**
         * 获取所有TopicViewMBean
         * @return
         * @throws Exception
         */
    	public Collection<TopicViewMBean> getTopics() throws Exception {
    		BrokerViewMBean broker = getBrokerAdmin();
    		if (broker == null) {
    			return Collections.EMPTY_LIST;
    		}
    		ObjectName[] queues = broker.getTopics();
    		return getManagedObjects(queues, TopicViewMBean.class);
    	}
        /**
         * 获取所有DurableSubscriptionViewMBean
         * @return
         * @throws Exception
         */
    	public Collection<DurableSubscriptionViewMBean> getDurableTopicSubscribers()
    			throws Exception {
    		BrokerViewMBean broker = getBrokerAdmin();
    		if (broker == null) {
    			return Collections.EMPTY_LIST;
    		}
    		ObjectName[] queues = broker.getDurableTopicSubscribers();
    		return getManagedObjects(queues, DurableSubscriptionViewMBean.class);
    	}
        /**
         * 获取所有DurableSubscriptionViewMBean
         * @return
         * @throws Exception
         */
    	public Collection<DurableSubscriptionViewMBean> getInactiveDurableTopicSubscribers()
    			throws Exception {
    		BrokerViewMBean broker = getBrokerAdmin();
    		if (broker == null) {
    			return Collections.EMPTY_LIST;
    		}
    		ObjectName[] queues = broker.getInactiveDurableTopicSubscribers();
    		return getManagedObjects(queues, DurableSubscriptionViewMBean.class);
    	}
        /**
         * 根据queueName获取queue相关的信息
         * @return
         * @throws Exception
         */
    	public QueueViewMBean getQueue(String name) throws Exception {
    		return (QueueViewMBean) getDestinationByName(getQueues(), name);
    	}
        /**
         * 根据topicName获取Topic相关的信息
         * @return
         * @throws Exception
         */
    	public TopicViewMBean getTopic(String name) throws Exception {
    		return (TopicViewMBean) getDestinationByName(getTopics(), name);
    	}
        /**
         * 获取DestinationViewMBean
         * @return
         * @throws Exception
         */
    	protected DestinationViewMBean getDestinationByName(
    			Collection<? extends DestinationViewMBean> collection, String name) {
    		Iterator<? extends DestinationViewMBean> iter = collection.iterator();
    		while (iter.hasNext()) {
    			DestinationViewMBean destinationViewMBean = iter.next();
    			if (name.equals(destinationViewMBean.getName())) {
    				return destinationViewMBean;
    			}
    		}
    		return null;
    	}
        /**
         * 获取所有Mananage
         * @return
         * @throws Exception
         */
    	@SuppressWarnings("unchecked")
    	protected <T> Collection<T> getManagedObjects(ObjectName[] names,
    			Class<T> type) throws Exception {
    		List<T> answer = new ArrayList<T>();
    		for (int i = 0; i < names.length; i++) {
    			ObjectName name = names[i];
    			T value = (T) newProxyInstance(name, type, true);
    			if (value != null) {
    				answer.add(value);
    			}
    		}
    		return answer;
    	}
        /**
         * 获取所有ConnectionViewMBean
         * @return
         * @throws Exception
         */
    	@SuppressWarnings("unchecked")
    	public Collection<ConnectionViewMBean> getConnections() throws Exception {
    		String brokerName = getBrokerName();
    		ObjectName query = new ObjectName("org.apache.activemq:BrokerName="
    				+ brokerName + ",Type=Connection,*");
    		Set<ObjectName> queryResult = queryNames(query, null);
    		return getManagedObjects(queryResult.toArray(new ObjectName[queryResult
    				.size()]), ConnectionViewMBean.class);
    	}
        /**
         * 获取所有Connections
         * @return
         * @throws Exception
         */
    	@SuppressWarnings("unchecked")
    	public Collection<String> getConnections(String connectorName)
    			throws Exception {
    		String brokerName = getBrokerName();
    		ObjectName query = new ObjectName("org.apache.activemq:BrokerName="
    				+ brokerName + ",Type=Connection,ConnectorName="
    				+ connectorName + ",*");
    		Set<ObjectName> queryResult = queryNames(query, null);
    		Collection<String> result = new ArrayList<String>(queryResult.size());
    		for (ObjectName on : queryResult) {
    			String name = StringUtils.replace(on.getKeyProperty("Connection"),
    					"_", ":");
    			result.add(name);
    		}
    		return result;
    	}
        /**
         * 获取所有ConnectionViewMBean
         * @return
         * @throws Exception
         */
    	@SuppressWarnings("unchecked")
    	public ConnectionViewMBean getConnection(String connectionName)
    			throws Exception {
    		connectionName = StringUtils.replace(connectionName, ":", "_");
    		String brokerName = getBrokerName();
    		ObjectName query = new ObjectName("org.apache.activemq:BrokerName="
    				+ brokerName + ",Type=Connection,*,Connection="
    				+ connectionName);
    		Set<ObjectName> queryResult = queryNames(query, null);
    		if (queryResult.size() == 0)
    			return null;
    		ObjectName objectName = queryResult.iterator().next();
    		return (ConnectionViewMBean) newProxyInstance(objectName,
    				ConnectionViewMBean.class, true);
    	}

    	@SuppressWarnings("unchecked")
    	public Collection<String> getConnectors() throws Exception {
    		String brokerName = getBrokerName();
    		ObjectName query = new ObjectName("org.apache.activemq:BrokerName="
    				+ brokerName + ",Type=Connector,*");
    		Set<ObjectName> queryResult = queryNames(query, null);
    		Collection<String> result = new ArrayList<String>(queryResult.size());
    		for (ObjectName on : queryResult)
    			result.add(on.getKeyProperty("ConnectorName"));
    		return result;
    	}

    	public ConnectorViewMBean getConnector(String name) throws Exception {
    		String brokerName = getBrokerName();
    		ObjectName objectName = new ObjectName(
    				"org.apache.activemq:BrokerName=" + brokerName
    						+ ",Type=Connector,ConnectorName=" + name);
    		return (ConnectorViewMBean) newProxyInstance(objectName,
    				ConnectorViewMBean.class, true);
    	}

    	@SuppressWarnings("unchecked")
    	public Collection<NetworkConnectorViewMBean> getNetworkConnectors()
    			throws Exception {
    		String brokerName = getBrokerName();
    		
    		ObjectName query = new ObjectName("org.apache.activemq:BrokerName="
    				+ brokerName + ",Type=NetworkConnector,*");
    		Set<ObjectName> queryResult = queryNames(query, null);
    		return getManagedObjects(queryResult.toArray(new ObjectName[queryResult
    				.size()]), NetworkConnectorViewMBean.class);
    	}

    	public Collection<NetworkBridgeViewMBean> getNetworkBridges()
    			throws Exception {
    		String brokerName = getBrokerName();
    		ObjectName query = new ObjectName("org.apache.activemq:BrokerName="
    				+ brokerName + ",Type=NetworkBridge,*");
    		Set<ObjectName> queryResult = queryNames(query, null);
    		return getManagedObjects(queryResult.toArray(new ObjectName[queryResult
    				.size()]), NetworkBridgeViewMBean.class);
    	}

    	@SuppressWarnings("unchecked")
    	public Collection<SubscriptionViewMBean> getQueueConsumers(String queueName)
    			throws Exception {
    		String brokerName = getBrokerName();
    		queueName = StringUtils.replace(queueName, "\"", "_");
    		ObjectName query = new ObjectName("org.apache.activemq:BrokerName="
    				+ brokerName
    				+ ",Type=Subscription,destinationType=Queue,destinationName="
    				+ queueName + ",*");
    		Set<ObjectName> queryResult = queryNames(query, null);
    		return getManagedObjects(queryResult.toArray(new ObjectName[queryResult
    				.size()]), SubscriptionViewMBean.class);
    	}

    	@SuppressWarnings("unchecked")
    	public Collection<SubscriptionViewMBean> getConsumersOnConnection(
    			String connectionName) throws Exception {
    		connectionName = StringUtils.replace(connectionName, ":", "_");
    		String brokerName = getBrokerName();
    		ObjectName query = new ObjectName("org.apache.activemq:BrokerName="
    				+ brokerName + ",Type=Subscription,clientId=" + connectionName
    				+ ",*");
    		Set<ObjectName> queryResult = queryNames(query, null);
    		return getManagedObjects(queryResult.toArray(new ObjectName[queryResult
    				.size()]), SubscriptionViewMBean.class);
    	}
        /**
         * 获取定时执行的队列的信息
         * @return
         * @throws Exception
         */
    	public JobSchedulerViewMBean getJobScheduler() throws Exception {
    		ObjectName name = getBrokerAdmin().getJMSJobScheduler();
    		return (JobSchedulerViewMBean) newProxyInstance(name,
    				JobSchedulerViewMBean.class, true);
    	}

    	public String getBrokerName() throws Exception {
    		return brokerService.getBrokerName();
    	}
    	/**
    	 * 获取Broker对象
    	 * @return
    	 * @throws Exception
    	 */
    	public Broker getBroker() throws Exception {
    		return brokerService.getBroker();
    	}
    	public ManagementContext getManagementContext() {
    		return brokerService.getManagementContext();
    	}

    	public ManagedRegionBroker getManagedBroker() throws Exception {
    		BrokerView adminView = brokerService.getAdminView();
    		if (adminView == null) {
    			return null;
    		}
    		return adminView.getBroker();
    	}

        public void purgeQueue(ActiveMQDestination destination) throws Exception {
            Set destinations = getManagedBroker().getQueueRegion().getDestinations(destination);
            for (Iterator i = destinations.iterator(); i.hasNext();) {
                Destination dest = (Destination) i.next();
                if (dest instanceof Queue) {
                    Queue regionQueue = (Queue) dest;
                    regionQueue.purge();
                }
            }
        }
        /**
         * 
         * @param name
         * @param query
         * @return
         * @throws Exception
         */
        public Set queryNames(ObjectName name, QueryExp query) throws Exception {
            return getManagementContext().queryNames(name, query);
        }
        /**
         * 通过JMX获取ActiveMQ各种信息
         * @param objectName
         * @param interfaceClass
         * @param notificationBroadcaster
         * @return
         */
        public Object newProxyInstance(ObjectName objectName, Class interfaceClass, boolean notificationBroadcaster) {
            return getManagementContext().newProxyInstance(objectName, interfaceClass, notificationBroadcaster);
        }
        /**
         * 监控内存信息
         * @throws Exception
         */
        public void monitorMermeryUsage() throws Exception{
        	SystemUsage proSystemUsage=brokerService.getProducerSystemUsage();
        	printSystemUsage(proSystemUsage);
        	SystemUsage syUage=brokerService.getSystemUsage();
        	printSystemUsage(syUage);
        	SystemUsage consumsyUage=brokerService.getConsumerSystemUsage();
        	printSystemUsage(consumsyUage);
        }
        
        /**
         * 打印内存信息
         * @param syUage
         */
        public void printSystemUsage(SystemUsage syUage){
        	String name=syUage.getName();
        	LOG.info("SystemUsage name ="+name);
        	MemoryUsage memeryUsage =syUage.getMemoryUsage();
        	LOG.info("memeryUsage PercentUsage name ="+memeryUsage.getPercentUsage());
        	LOG.info("memeryUsage Limit name ="+memeryUsage.getLimit());
        	LOG.info("memeryUsage Usage name ="+memeryUsage.getUsage());
        	TempUsage tempUsage =syUage.getTempUsage();
        	LOG.info("tempUsage PercentUsage name ="+tempUsage.getPercentUsage());
        	LOG.info("tempUsage Limit name ="+tempUsage.getLimit());
        	LOG.info("tempUsage Usage name ="+tempUsage.getUsage());
        	StoreUsage storeUsage=syUage.getStoreUsage();
        	LOG.info("storeUsage PercentUsage name ="+storeUsage.getPercentUsage());
        	LOG.info("storeUsage Limit name ="+storeUsage.getLimit());
        	LOG.info("storeUsage Usage name ="+storeUsage.getUsage());
        }
	    /**
	     * 监控消息的方法
	     * @throws Exception
	     */
	    public void monitorQueueAndTopic() throws Exception{
	    	LOG.info("==========Connection =================");
	    	Collection<ConnectionViewMBean> conVBean=getConnections();
	    	for (ConnectionViewMBean bean : conVBean) {
	    		LOG.info("remoteAddress:"+bean.getRemoteAddress());
	    		LOG.info("isActive:"+bean.isActive());
	    		LOG.info("isConnected:"+bean.isConnected());
			}
	    	LOG.info("=============Topic =================");
	    	Collection<TopicViewMBean>  topicVBean=getTopics();
	    	for (TopicViewMBean topicbean : topicVBean) {
	    		LOG.info("beanName ="+topicbean.getName());
	    		LOG.info("ConsumerCount ="+topicbean.getConsumerCount());
	    		LOG.info("DequeueCount ="+topicbean.getDequeueCount());
	    		LOG.info("EnqueueCount ="+topicbean.getEnqueueCount());
	    		LOG.info("DispatchCount ="+topicbean.getDispatchCount());
	    		LOG.info("ExpiredCount ="+topicbean.getExpiredCount());
	    		LOG.info("MaxEnqueueTime ="+topicbean.getMaxEnqueueTime());
	    		LOG.info("ProducerCount ="+topicbean.getProducerCount());
	    		LOG.info("MemoryPercentUsage ="+topicbean.getMemoryPercentUsage());
	    		LOG.info("MemoryLimit ="+topicbean.getMemoryLimit());
			}
	    	LOG.info("============Queue===================");
	    	Collection<QueueViewMBean> queuqVBeanList=getQueues();
	      	for (QueueViewMBean queuebean : queuqVBeanList) {
	    		LOG.info(" queue beanName ="+queuebean.getName());
	    		LOG.info("ConsumerCount ="+queuebean.getConsumerCount());
	    		LOG.info("DequeueCount ="+queuebean.getDequeueCount());
	    		LOG.info("EnqueueCount ="+queuebean.getEnqueueCount());
	    		LOG.info("DispatchCount ="+queuebean.getDispatchCount());
	    		LOG.info("ExpiredCount ="+queuebean.getExpiredCount());
	    		LOG.info("MaxEnqueueTime ="+queuebean.getMaxEnqueueTime());
	    		LOG.info("ProducerCount ="+queuebean.getProducerCount());
	    		LOG.info("MemoryPercentUsage ="+queuebean.getMemoryPercentUsage());
	    		LOG.info("MemoryLimit ="+queuebean.getMemoryLimit());
			}
	    }
	    
	    public void test() throws Exception{
	       	//获取初始化信息
	    	init();
	    	for (int i = 0; i < 10; i++) {
	    		sendTopic(i);
			}
	    	for (int i = 0; i < 10; i++) {
	    		sendPS(i);
			}
            monitorQueueAndTopic();	    	
	    	Thread.sleep(5000);
            receiveTopic();
	    	receivePS();
	    }
	    /**
	     * P2P发送方式
	     * @throws JMSException
	     */
	    public void sendTopic(int i) throws JMSException{
	    	Session session=connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
	    	Destination topic=session.createQueue("activemq.queue"+i);
	    	MessageProducer productor=(MessageProducer) session.createProducer(topic);
	    	TextMessage txtMessage =session.createTextMessage();
	    	txtMessage.setText("this is a topic message "+i);
	    	productor.send(txtMessage);
	    }
	    /**
	     * Sub/Pub发送方式
	     * @throws JMSException
	     */
	    public void sendPS(int i) throws JMSException{
	    	Session session=connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
	    	Destination topic=session.createTopic("activemq.topic"+i);
	    	MessageProducer productor=(MessageProducer) session.createProducer(topic);
	    	TextMessage txtMessage =session.createTextMessage();
	    	txtMessage.setText("this is a topic message "+i);
	    	productor.send(txtMessage);
	    }
	    
	    /**
	     * P2P接受方式
	     * @throws JMSException
	     */
	    public void receiveTopic() throws JMSException{
	    	Session session=connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
	    	Destination topic=session.createQueue("activemq.queue");
	    	
	    	MessageConsumer consumer=(MessageConsumer) session.createConsumer(topic);
	    	TextMessage txtMessage =(TextMessage)consumer.receive();
	    	System.out.println("txtMessage ="+txtMessage.getText());
	    }
	    /**
	     * Sub/Pub接受方式
	     * @throws JMSException
	     */
	    public void receivePS() throws JMSException{
	       	Session session=connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
	    	Destination topic=session.createQueue("activemq.topic");
	    	
	    	MessageConsumer consumer=(MessageConsumer) session.createConsumer(topic);
	    	TextMessage txtMessage =(TextMessage)consumer.receive();
	    	System.out.println("txtMessage ="+txtMessage.getText());
	    }
	    
	    
	    /**
	     * 初始化消息的方法
	     * @throws Exception
	     */
	    protected void init() throws Exception {
	        if (brokerService == null) {
	            brokerService = createBroker();
	        }
	        ActiveMQConnectionFactory factory = createConnectionFactory();
	        connection = factory.createConnection();
	        //添加Connection 的状态监控的方法
	        monitorConnection(connection);
	        //启动连接
	        connection.start();
	    }
	    
	    /**
	     * 监控台ActiveMQConnection的状态的方法
	     * @param connection
	     */
	    public void monitorConnection(Connection connection){
	    	ActiveMQConnection activemqconnection =(ActiveMQConnection)connection;
	    	//添加ActiveMQConnection的监听类
	    	activemqconnection.addTransportListener(new TransportListener(){

				public void onCommand(Object object) {
					LOG.info("onCommand  object "+object);
					
				}

				public void onException(IOException ex) {
					LOG.info("onException ="+ex.getMessage());
				}

				public void transportInterupted() {
					LOG.info("transportInterupted =");
				}

				public void transportResumed() {
					LOG.info("transportResumed .........");					
				}
	    		
	    	});
	    }

	    protected void destory() throws Exception {
	        connection.close();
	        if (brokerService != null) {
	            brokerService.stop();
	        }
	    }
        /**
         * 创建ActiveMQConnectionFactory
         * @return
         * @throws Exception
         */
	    protected ActiveMQConnectionFactory createConnectionFactory()
	            throws Exception {
	        ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(
	                bindAddress);
	        return cf;
	    }

	    /***
	     * 创建一个Broker监听进程
	     * @return
	     * @throws Exception
	     */
	    protected BrokerService createBroker() throws Exception {
	    	//创建BrokerService对象
	        BrokerService answer = new BrokerService();
	        //配置监听相关的信息
	        configureBroker(answer);
	        //启动Broker的启动
	        answer.start();
	        
	        return answer;
	    }

	    /**
	     * 配置Broker
	     * @param answer
	     * @throws Exception
	     */
	    protected void configureBroker(BrokerService answer) throws Exception {
	        //创建持久化信息
	    	answer.setPersistent(false);
	    	//设置采用JMX管理
	    	answer.setUseJmx(true);
	        ConstantPendingMessageLimitStrategy strategy = new ConstantPendingMessageLimitStrategy();
	        strategy.setLimit(10);
	        PolicyEntry tempQueueEntry = createPolicyEntry(strategy);
	        tempQueueEntry.setTempQueue(true);
	        PolicyEntry tempTopicEntry = createPolicyEntry(strategy);
	        tempTopicEntry.setTempTopic(true);
	        PolicyMap pMap = new PolicyMap();
	        final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();
	        policyEntries.add(tempQueueEntry);
	        policyEntries.add(tempTopicEntry);
	        pMap.setPolicyEntries(policyEntries);
	        answer.setDestinationPolicy(pMap);
	        //绑定url
	        answer.addConnector(bindAddress);
	        answer.setDeleteAllMessagesOnStartup(true);
	    }
        /**
         * 创建一个配置策略
         * @param strategy
         * @return
         */
	    private PolicyEntry createPolicyEntry(ConstantPendingMessageLimitStrategy strategy) {
	        PolicyEntry policy = new PolicyEntry();
	        policy.setAdvisdoryForFastProducers(true);
	        policy.setAdvisoryForConsumed(true);
	        policy.setAdvisoryForDelivery(true);
	        policy.setAdvisoryForDiscardingMessages(true);
	        policy.setAdvisoryForSlowConsumers(true);
	        policy.setAdvisoryWhenFull(true);
	        policy.setProducerFlowControl(false);
	        policy.setPendingMessageLimitStrategy(strategy);
	        return policy;
	    }
	    
	    public void object2string(Object object ){
	    	ToStringBuilder.reflectionToString(object, ToStringStyle.MULTI_LINE_STYLE);
	    }
	    
}

package easyway.app.activemq.demo.monitors;

public class ActiveMQMonitorTest {
	public static void main(String[] args) throws Exception {
		ActiveMQMonitor monitor=new ActiveMQMonitor();
		monitor.test();
		
	}
}

分享到：

ActiveMQ的JMX监控使用 | activemq消息队列的大小配置的配置

2011-06-08 13:56
浏览 16103
评论(17)
分类:编程语言
查看更多

17 楼 ZY199266 2018-01-31

获取不到任何消息信息，请问这是什么原因呢？

16 楼 xiaoyao霄 2018-01-29

DestinationSourceMonitor 报错应该导入那个包

15 楼 zzd0058 2017-07-26

你好我想知道getConnections为啥是空的

14 楼 longgangbai 2012-07-16

myfwz 写道

你有没有做过类似生产的速度与消费者（一或多个）的速度对比的工具。

没有响应的工具

13 楼 longgangbai 2012-07-16

myfwz 写道

1099这个端口在哪个配置文件里面？需要手动执行activemq-admin start来开启这个JMs监控端口是吗？

actviemq的admin管理端好像有一个bat文件可以编辑的。你修改配置一下

12 楼 myfwz 2012-07-10

你有没有做过类似生产的速度与消费者（一或多个）的速度对比的工具。

11 楼 myfwz 2012-07-10

1099这个端口在哪个配置文件里面？需要手动执行activemq-admin start来开启这个JMs监控端口是吗？

10 楼 longgangbai 2012-07-10

myfwz 写道

我在本地运行的时候，不管是连我自己的MQ，还是远程的MQ，都没法运行起来。
本地我用netstat看过了，远程也一样使用netsat -ntl看过了，没有起1099这个端口。
另外：1099这个端口是ActiveMQ启动的时候拉起来的，还是监控程序设置的一个端口？

activemq启动时候的使用的端口。
默认的是tcp://localhost:61616
1099为jms监控端口：
activemq-admin stop --jmxurl service:jmx:rmi:///jndi/rmi://remotehost:1099/jmxrmi --all

9 楼 longgangbai 2012-07-10

myfwz 写道

呵呵，非常感谢你的回复，只是我想了解一下，你在运行这个类的时候是连接本地MQ还是远程的MQ，是否需要在MQ上面做相应的设置，如果有的话，麻烦提供相应的配置文件给我学习下。

activemq安装之后有一个web版你发送多个消息，不要点击接受，之后，开启这个程序运行。我当时是这样测试的。

8 楼 myfwz 2012-07-10

7 楼 myfwz 2012-07-10

6 楼 longgangbai 2012-07-10

myfwz 写道

1099这个端口是ActiveMQ提供出来的吗？我的电脑重启后，再去连接也会报这个错呢。

java.rmi.server.ExportException: Port already in use: 1099; nested exception is:
java.net.BindException: Address already in use: JVM_Bind
这句话的中文意思，端口1099已经使用，JVM_Bind已经使用该地址。

5 楼 myfwz 2012-07-09

1099这个端口是ActiveMQ提供出来的吗？我的电脑重启后，再去连接也会报这个错呢。

4 楼 longgangbai 2012-07-04

myfwz 写道

你好，我在本地运行一个ActiveMQ，再运行这个程序报了以下异常：

(1 ms) [main] DEBUG: org.apache.activemq.broker.jmx.ManagementContext#createConnector : Failed to create local registry
java.rmi.server.ExportException: Port already in use: 1099; nested exception is:
java.net.BindException: Address already in use: JVM_Bind
at sun.rmi.transport.tcp.TCPTransport.listen(TCPTransport.java:310)
at sun.rmi.transport.tcp.TCPTransport.exportObject(TCPTransport.java:218)
at sun.rmi.transport.tcp.TCPEndpoint.exportObject(TCPEndpoint.java:393)
at sun.rmi.transport.LiveRef.exportObject(LiveRef.java:129)
at sun.rmi.server.UnicastServerRef.exportObject(UnicastServerRef.java:190)
at sun.rmi.registry.RegistryImpl.setup(RegistryImpl.java:92)
at sun.rmi.registry.RegistryImpl.<init>(RegistryImpl.java:68)
at java.rmi.registry.LocateRegistry.createRegistry(LocateRegistry.java:222)

JMS对应的本地注册的本地端口意见使用，netstat -a | find "1099" 查看那个程序占用，或者重启机器，再试！

3 楼 myfwz 2012-06-25

2 楼 longgangbai 2012-05-25

962685987962685987 写道

你好，我想问一下，我在本地搭建的 activeMQ WEB 界面中发送了几个消息（queue），但是通过上面的程序却无法获取消息信息，即是说：getQueues() 获取不到任何消息信息，请问这是什么原因呢？

你可能要检查你的activemq配置是否支持jmx，如果不支持是监控不到的，上面的demo是基于jmx实现的，请检查相关的配置。

1 楼 962685987962685987 2012-05-24

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论