ActiveMQ(三) 转

package pfs.y2017.m11.mq.activemq.demo03;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer {

    // ActiveMq 的默认用户名
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    // ActiveMq 的默认登录密码
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    // ActiveMQ 的链接地址
    private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;

    ConnectionFactory factory;

    Connection connection;

    Session session;

    MessageProducer producer;

    Destination[] destinations;

    public Consumer() throws JMSException {
        factory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);
        connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    } 

    public static void main(String[] args) throws JMSException {
        Consumer consumer = new Consumer();
        String[] arg= {"aa","bb"};

        for (String stock : arg) {
        Destination destination = consumer.getSession().createTopic("STOCKS." + stock);
        MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);
        messageConsumer.setMessageListener(new Listener());
        }
    }  

    public Session getSession() {
        return session;
    }  

}

package pfs.y2017.m11.mq.activemq.demo03;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQMapMessage;

public class Publisher {

    // ActiveMq 的默认用户名
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    // ActiveMq 的默认登录密码
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    // ActiveMQ 的链接地址
    private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;

    ConnectionFactory factory;

    Connection connection;

    Session session;

    MessageProducer producer;

    Destination[] destinations;

    public Publisher() throws JMSException {
        factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEN_URL);
        connection = factory.createConnection();
        try {
            connection.start();
        } catch (JMSException jmse) {
            connection.close();
            throw jmse;
        }
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = session.createProducer(null);
    }

    protected void setTopics(String[] stocks) throws JMSException {
        destinations = new Destination[stocks.length];
        for (int i = 0; i < stocks.length; i++) {
            destinations[i] = session.createTopic("STOCKS." + stocks[i]);
        }
    }

    protected void sendMessage(String[] stocks) throws JMSException {
        for (int i = 0; i < stocks.length; i++) {
            Message message = createStockMessage(stocks[i], session);
            System.out.println("Sending: " + ((ActiveMQMapMessage) message).getContentMap() + " on destination: "
                    + destinations[i]);
            producer.send(destinations[i], message);
        }
    }

    protected Message createStockMessage(String stock, Session session) throws JMSException {
        MapMessage message = session.createMapMessage();
        message.setString("stock", stock);
        message.setDouble("price", 1.00);
        message.setDouble("offer", 0.01);
        message.setBoolean("up", true);

        return message;
    }

    public static void main(String[] args) throws JMSException {
//        if (args.length < 1)
//            throw new IllegalArgumentException();

        // Create publisher
        Publisher publisher = new Publisher();

        String[] arg= {"aa","bb"};
        // Set topics
        publisher.setTopics(arg);

        for (int i = 0; i < 10; i++) {
            publisher.sendMessage(arg);
            System.out.println("Publisher ‘" + i + " price messages");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // Close all resources
        publisher.close();
    }

    public void close() throws JMSException {
        if (connection != null) {
            connection.close();
        }
    }
}

package pfs.y2017.m11.mq.activemq.demo03;

import java.text.DecimalFormat;

import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;

public class Listener implements MessageListener {

    public 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();
        }
    }

}