activemq 怎么用编程方式得到 queue上未处理的消息？

0 0

activemq 怎么用编程方式得到 queue上未处理的消息？5

JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi");
JMXConnector jmxc = JMXConnectorFactory.connect(url);
MBeanServerConnection conn = jmxc.getMBeanServerConnection();
ObjectName activeMQ = new ObjectName("org.apache.activemq:BrokerName=localhost,Type=Broker");
BrokerViewMBean mbean = (BrokerViewMBean) MBeanServerInvocationHandler.newProxyInstance(conn, activeMQ,BrokerViewMBean.class, true);

for (ObjectName name : mbean.getQueues()) {
//	List<?> mlist = (List<?>) conn.invoke(name, "browseMessages", new Object[] {}, new String[] {});
	CompositeData[] messages = (CompositeData[]) conn.invoke(name, "browse", new Object[] {}, new String[] {});
	for( CompositeData message :messages){
		System.out.println(message.get("Text"));
	}
}

我目前是用这种方式查询queue的。但只有消息为 TextMessage时，message.get("Text")才能取到值。
我猜是因为QueueViewMBean的browse方法得到的是CompositeData[]，和消息差距较大。

查看api发现的browseMessages

List<?>	browseMessages() 
          Browses the current destination returning a list of messages

看起来这个更适合，但是一用就保错了。

Exception in thread "main" java.rmi.UnmarshalException: error unmarshalling return; nested exception is: 
	java.io.WriteAbortedException: writing aborted; java.io.NotSerializableException: org.apache.activemq.command.ActiveMQObjectMessage
	at sun.rmi.server.UnicastRef.invoke(Unknown Source)
	at com.sun.jmx.remote.internal.PRef.invoke(Unknown Source)
	at javax.management.remote.rmi.RMIConnectionImpl_Stub.invoke(Unknown Source)
	at javax.management.remote.rmi.RMIConnector$RemoteMBeanServerConnection.invoke(Unknown Source)
	at org.DestinationSourceM.main(DestinationSourceM.java:49)
Caused by: java.io.WriteAbortedException: writing aborted; java.io.NotSerializableException: org.apache.activemq.command.ActiveMQObjectMessage
	at java.io.ObjectInputStream.readObject0(Unknown Source)
	at java.io.ObjectInputStream.readObject(Unknown Source)
	at java.util.ArrayList.readObject(Unknown Source)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
	at java.lang.reflect.Method.invoke(Unknown Source)
	at java.io.ObjectStreamClass.invokeReadObject(Unknown Source)
	at java.io.ObjectInputStream.readSerialData(Unknown Source)
	at java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
	at java.io.ObjectInputStream.readObject0(Unknown Source)
	at java.io.ObjectInputStream.readObject(Unknown Source)
	at sun.rmi.server.UnicastRef.unmarshalValue(Unknown Source)
	... 5 more
Caused by: java.io.NotSerializableException: org.apache.activemq.command.ActiveMQObjectMessage
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1164)
	at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:330)
	at java.util.ArrayList.writeObject(ArrayList.java:570)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
	at java.lang.reflect.Method.invoke(Method.java:597)
	at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:945)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1469)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1400)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1158)
	at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:330)
	at sun.rmi.server.UnicastRef.marshalValue(UnicastRef.java:274)
	at sun.rmi.server.UnicastServerRef.dispatch(UnicastServerRef.java:315)
	at sun.rmi.transport.Transport$1.run(Transport.java:159)
	at java.security.AccessController.doPrivileged(Native Method)
	at sun.rmi.transport.Transport.serviceCall(Transport.java:155)
	at sun.rmi.transport.tcp.TCPTransport.handleMessages(TCPTransport.java:535)
	at sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.run0(TCPTransport.java:790)
	at sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.run(TCPTransport.java:649)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
	at java.lang.Thread.run(Thread.java:662)