import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

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

public class Producer {

    // ActiveMq的地址、用户名、密码，这里都是默认的。
    private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    private static String user = ActiveMQConnection.DEFAULT_USER;
    private static String password = ActiveMQConnection.DEFAULT_PASSWORD;

    // Topic的名称
    private static String topicName = "userSyncTopic";

    public static void main(String[] args) {

        Connection connection = null;

        try {
            // 创建一个连接
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                    user, password, url);
            connection = connectionFactory.createConnection();
            connection.start();

            // 创建一个session
            Session session = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);

            // 取得Topic，没有的话，会自动创建
            Topic topic = session.createTopic(topicName);

            // 创建一个消息生产者对象
            MessageProducer producer = session.createProducer(topic);

            // 设置消息持久化存储
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);

            // 创建一个消息对象，这是用的是普通的TextMessage
            TextMessage message = session.createTextMessage("你好，这是个简单的消息文本");

            // 消息发送
            producer.send(message);

            // 关一切
            connection.stop();
            connection.close();

        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

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

public class Consumer {

    // ActiveMq的地址、用户名、密码，这里都是默认的。
    private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    private static String user = ActiveMQConnection.DEFAULT_USER;
    private static String password = ActiveMQConnection.DEFAULT_PASSWORD;

    // Topic的名称
    private static String topicName = "userSyncTopic";

    public static void main(String[] args) {

        Connection connection = null;

        try {
            // 创建一个连接
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                    user, password, url);
            connection = connectionFactory.createConnection();
            connection.setClientID("aaa"); // 设置客户端ID，持久订阅必须的。
            connection.start();

            // 创建一个session
            Session session = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);

            // 取得Topic，没有的话，会自动创建
            Topic topic = session.createTopic(topicName);

            // 创建持久订阅，aaa是订户名称，和clientID可以不同
            MessageConsumer consumer = session.createDurableSubscriber(topic,
                    "aaa");

            // 这样，是普通订阅，消费者没启动的话，期间的消息都收不到。
            // MessageConsumer consumer = session.createConsumer(destination);

            // 循环接收消息
            long a = System.currentTimeMillis();
            long b = System.currentTimeMillis();
            while (b - a < 30000) {

                // 接收消息，不等待
                Message message = consumer.receiveNoWait();
                if (message != null) {

                    // 实际应用时，根据需要转换类型，这里是普通文本，所以转为TextMessage
                    System.out.println("收到的message是:"
                            + ((TextMessage) message).getText());
                } else {
                    System.out.println("no messsage.");
                }

                // 休息1秒
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(".." + (b - a) / 1000);
                b = System.currentTimeMillis();
            }

            // 关一切
            connection.close();

        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

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

public class Consumer2 {

    // ActiveMq的地址、用户名、密码，这里都是默认的。
    private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    private static String user = ActiveMQConnection.DEFAULT_USER;
    private static String password = ActiveMQConnection.DEFAULT_PASSWORD;

    // Topic的名称
    private static String topicName = "userSyncTopic";

    public static void main(String[] args) {

        Connection connection = null;

        try {
            // 创建一个连接
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                    user, password, url);
            connection = connectionFactory.createConnection();
            connection.setClientID("bbb"); // 设置客户端ID，持久订阅必须的。
            connection.start();

            // 创建一个session
            Session session = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);

            // 取得Topic，没有的话，会自动创建
            Topic topic = session.createTopic(topicName);

            // 创建持久订阅，bbb是订户名称，和clientID可以不同
            MessageConsumer consumer = session.createDurableSubscriber(topic,
                    "bbb");

            // 这样，是普通订阅，消费者没启动的话，期间的消息都收不到。
            // MessageConsumer consumer = session.createConsumer(destination);

            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {

                    if (message instanceof TextMessage) {
                        try {
                            System.out.println("收到的message是:"
                                    + ((TextMessage) message).getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });

            // 等待30秒结束
            long a = System.currentTimeMillis();
            long b = System.currentTimeMillis();
            while (b - a < 30000) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(".." + (b - a) / 1000);
                b = System.currentTimeMillis();
            }
            // 关一切
            connection.close();

        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}