import javax.jms.Connection;public class Consumer {private static String brokerURL = "tcp://localhost:61616";private static transient ConnectionFactory factory;private transient Connection connection;private transient Session session;private String jobs[] = new String[]{"suspend", "delete"};public Consumer() throws JMSException {factory = new ActiveMQConnectionFactory(brokerURL);connection = factory.createConnection();connection.start();session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);}public void close() throws JMSException {if (connection != null) {connection.close();}}    public static void main(String[] args) throws JMSException {Consumer consumer = new Consumer();for (String job : consumer.jobs) {Destination destination = consumer.getSession().createQueue("JOBS." + job);MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);messageConsumer.setMessageListener(new Listener(job));}}public Session getSession() {return session;}}

import javax.jms.Message;public class Listener implements MessageListener {private String job;public Listener(String job) {this.job = job;}public void onMessage(Message message) {try {//do something hereSystem.out.println(job + " id:" + ((ObjectMessage)message).getObject());} catch (Exception e) {e.printStackTrace();}}}

import javax.jms.Connection;public class Publisher {private static String brokerURL = "tcp://localhost:61616";private static transient ConnectionFactory factory;private transient Connection connection;private transient Session session;private transient MessageProducer producer;private static int count = 10;private static int total;private static int id = 1000000;private String jobs[] = new String[]{"suspend", "delete"};public Publisher() throws JMSException {factory = new ActiveMQConnectionFactory(brokerURL);connection = factory.createConnection();connection.start();session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);producer = session.createProducer(null);}    public void close() throws JMSException {if (connection != null) {connection.close();}}    public static void main(String[] args) throws JMSException {Publisher publisher = new Publisher();while (total < 1000) {for (int i = 0; i < count; i++) {publisher.sendMessage();}total += count;System.out.println("Published '" + count + "' of '" + total + "' job messages");try {Thread.sleep(1000);} catch (InterruptedException x) {}}publisher.close();}public void sendMessage() throws JMSException {int idx = 0;while (true) {idx = (int)Math.round(jobs.length * Math.random());if (idx < jobs.length) {break;}}String job = jobs[idx];Destination destination = session.createQueue("JOBS." + job);Message message = session.createObjectMessage(id++);System.out.println("Sending: id: " + ((ObjectMessage)message).getObject() + " on queue: " + destination);producer.send(destination, message);}    }

import javax.jms.Connection;public class Consumer {private static String brokerURL = "tcp://localhost:61616";private static transient ConnectionFactory factory;private transient Connection connection;private transient Session session;public Consumer() throws JMSException {factory = new ActiveMQConnectionFactory(brokerURL);connection = factory.createConnection();connection.start();/*通知方式                    效果DUPS_OK_ACKNOWLEDGE         session延迟通知。如果JMS服务器宕机，会造成重复消息的情况。程序必须保证处理重复消息而不引起程序逻辑的混乱。AUTO_ACKNOWLEDGE             当receive或MessageListener方法成功返回后自动通知。CLIENT_ACKNOWLEDGE            客户端调用消息的acknowledge方法通知*/session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);}public void close() throws JMSException {if (connection != null) {connection.close();}}    public static void main(String[] args) throws JMSException {Consumer consumer = new Consumer();for (String stock : args) {Destination destination = consumer.getSession().createTopic("STOCKS." + stock);// A client uses a MessageConsumer object to receive messages from a destinationMessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);// 注册一个Message Listener，实现异步messageConsumer.setMessageListener(new Listener());}}public Session getSession() {return session;}}

import java.text.DecimalFormat;
//A MessageListener object is used to receive asynchronously delivered messages
public class Listener implements MessageListener {// 处理messagepublic void onMessage(Message message) {try {MapMessage map = (MapMessage)message;String stock = map.getString("stock");double price = map.getDouble("price");double offer = map.getDouble("offer");boolean up = map.getBoolean("up");DecimalFormat df = new DecimalFormat( "#,###,###,##0.00" );System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t" + (up?"up":"down"));} catch (Exception e) {e.printStackTrace();}}}

import java.util.Hashtable;public class Publisher {protected int MAX_DELTA_PERCENT = 1;protected Map<String, Double> LAST_PRICES = new Hashtable<String, Double>();protected static int count = 10;protected static int total;protected static String brokerURL = "tcp://localhost:61616";protected static transient ConnectionFactory factory;protected transient Connection connection;protected transient Session session;protected transient MessageProducer producer;public Publisher() throws JMSException {factory = new ActiveMQConnectionFactory(brokerURL);connection = factory.createConnection();try {connection.start();} catch (JMSException jmse) {connection.close();throw jmse;}session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);producer = session.createProducer(null);}public void close() throws JMSException {if (connection != null) {connection.close();}}public static void main(String[] args) throws JMSException {Publisher publisher = new Publisher();while (total < 1000) {for (int i = 0; i < count; i++) {publisher.sendMessage(args);}total += count;System.out.println("Published '" + count + "' of '" + total + "' price messages");try {Thread.sleep(1000);} catch (InterruptedException x) {}}publisher.close();}protected void sendMessage(String[] stocks) throws JMSException {int idx = 0;while (true) {idx = (int)Math.round(stocks.length * Math.random());if (idx < stocks.length) {break;}}String stock = stocks[idx];// Topic和Queue都继承与Destination，也就这两种模式Destination destination = session.createTopic("STOCKS." + stock);Message message = createStockMessage(stock, session);System.out.println("Sending: " + ((ActiveMQMapMessage)message).getContentMap() + " on destination: " + destination);producer.send(destination, message);}protected Message createStockMessage(String stock, Session session) throws JMSException {Double value = LAST_PRICES.get(stock);if (value == null) {value = new Double(Math.random() * 100);}// lets mutate the value by some percentagedouble oldPrice = value.doubleValue();value = new Double(mutatePrice(oldPrice));LAST_PRICES.put(stock, value);double price = value.doubleValue();double offer = price * 1.001;boolean up = (price > oldPrice);MapMessage message = session.createMapMessage();message.setString("stock", stock);message.setDouble("price", price);message.setDouble("offer", offer);message.setBoolean("up", up);return message;}protected double mutatePrice(double price) {double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT;return price * (100 + percentChange) / 100;}}