activeMQ的一个简单的demo

package com.daiy.cons; import java.io.BufferedWriter; import java.io.File; import java.io.FileWriter; import java.io.IOException; import java.util.Calendar; import javax.jms.Connection; import javax.jms.ExceptionListener; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.Session; import javax.jms.ObjectMessage; import javax.jms.TextMessage; import javax.jms.Topic; import com.daiy.Student; import com.daiy.send.ToolSupport; import com.daiy.test.TestXML; /** * 接收类，消费者模型 */ public class SmsMQRecv extends ToolSupport implements MessageListener, ExceptionListener { protected int count = 0; protected int dumpCount = 10; protected boolean verbose = true; protected int maximumMessages = 0; private boolean pauseBeforeShutdown; private boolean running; private Session session; private long sleepTime = 0; /**构造器 * */ public SmsMQRecv(){ } public static void main(String[] args) { SmsMQRecv tool = new SmsMQRecv(); // LoadIni ini =new LoadIni(); //装载配置文件 // ini.setFilename("web" + File.separator + "smsmq.ini"); try{ tool.clientID = null;//ini.readINI("clientID"); tool.durable = false;//ini.readINI("durable").equals("true") ? true : false; // tool.subject = "SMS.MQ";//ini.readINI("subject"); tool.subject = "TEST.FOO";//ini.readINI("subject"); tool.url = "tcp://localhost:61616";//ini.readINI("url"); tool.topic = false;//ini.readINI("topic").equals("true")? true: false; tool.sleepTime = 0;//Integer.parseInt(ini.readINI("sleepTime")); }catch(Exception e){ e.printStackTrace(); } if (args.length > 4) { tool.maximumMessages = Integer.parseInt(args[4]); } //if (args.length > 6) { // tool.transacted = "true".equals(args[6]); //} tool.run(); } public void run() { try { running = true; System.out.println("连接到服务器: " + url); System.out.println("Consuming " + (topic ? "topic" : "queue") + ": " + subject); System.out.println("Using " + (durable ? "durable" : "non-durable") + " subscription"); Connection connection = createConnection();//建立一个消息池连接 connection.setExceptionListener(this);//设置ExceptionListener为this ,因为this implements ExceptionListener session = createSession(connection); MessageConsumer consumer = null;//建立一个消息消费者 if (durable && topic) { consumer = session.createDurableSubscriber((Topic) destination, consumerName); } else { consumer = session.createConsumer(destination); } if (maximumMessages <= 0) { consumer.setMessageListener(this); } if (maximumMessages > 0) { consumeMessagesAndClose(connection, session, consumer); } } catch (Exception e) { System.out.println("Caught: " + e); e.printStackTrace(); } } /**MessageListener 的abstract 接口 * @param Message 从消息池接收到的消息*/ public void onMessage(Message message) { try { if (message instanceof TextMessage) { // ObjectMessage Msg = ((ObjectMessage) message); // String temp = Msg.getStringProperty("pro"); // System.out.println(temp); // Student sd = (Student)(Msg.getObject()); // System.out.println(sd.getName()+"----"+sd.getAddress()); // if (verbose) System.out.println("Received: " + sd.getAge()); TextMessage txtMsg = (TextMessage) message; String msg = txtMsg.getText(); String path = writeFile(msg); System.out.println("Consumer:->收到的消息: " + msg); TestXML xml = new TestXML(); xml.init(); xml.parserXml(path); } else { if (verbose) System.out.println("Received: " + message); } if (transacted) { session.commit(); } } catch (JMSException e) { System.out.println("Caught: " + e); e.printStackTrace(); } finally { if (sleepTime > 0) { try { Thread.sleep(sleepTime); } catch (InterruptedException e) { e.printStackTrace(); } } } } /**ExceptionListener的abstract 接口 *@param JMSException 接收到的JMS异常 */ synchronized public void onException(JMSException ex) { System.out.println("JMS Exception occured. Shutting down client."); running = false; } /**检测实例是否正在运行*/ synchronized boolean isRunning() { return running; } protected void consumeMessagesAndClose(Connection connection, Session session, MessageConsumer consumer) throws JMSException, IOException { System.out.println("系统等待,直到接收到 " + maximumMessages + " 条消息再退出"); while( isRunning()){ Message message = consumer.receive(); onMessage(message); } /* for (int i = 0; i < maximumMessages && isRunning();) { Message message = consumer.receive(10); if (message != null) { i++; onMessage(message); } }*/ System.out.println("Closing connection"); consumer.close(); session.close(); connection.close(); if (pauseBeforeShutdown) { System.out.println("Press return to shut down"); System.in.read(); } } /** * * @param res 文件内容 */ private String writeFile(String res){ BufferedWriter bw = null; String temp = ""; try { Calendar time = Calendar.getInstance(); String fileName = time.getTime().getTime()+".xml"; File file = new File("src/xmlFile/"+fileName); bw = new BufferedWriter(new FileWriter(file)); bw.write(res); bw.close(); temp = file.getPath(); } catch (Exception e) { temp = ""; try { if(bw != null){ bw.close(); } } catch (IOException e1) { temp = ""; e1.printStackTrace(); } e.printStackTrace(); } return temp; } }