amq 小程序

package learn.amq;

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

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * Date: 2016-12-27
 *
 * @author
 */
public class MsgProducer {

    private static final String BROKER_URL = "failover://tcp://127.0.0.1:61616";

    public static void main(String[] args) throws JMSException, InterruptedException {
        //建立鏈接工廠
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        //得到鏈接
        Connection conn = connectionFactory.createConnection();
        //start
        conn.start();

        //建立Session，此方法第一個參數表示會話是否在事務中執行，第二個參數設定會話的應答模式
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

        //建立隊列
        Destination dest = session.createQueue("test-queue");
        //建立消息生產者
        MessageProducer producer = session.createProducer(dest);

        for (int i = 0; i < 100; i++) {
            //初始化一個mq消息
            TextMessage message = session.createTextMessage("這是第 " + i + " 條消息！");
            //發送消息
            producer.send(message);
            System.out.println("send message:消息" + i);
            //暫停3秒
            Thread.sleep(100);
        }

        //關閉mq鏈接
        conn.close();
    }

}


import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
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 org.apache.activemq.ActiveMQConnectionFactory;

/**
 * Date: 2016-12-27
 *
 * @author
 */
public class MsgConsumer implements MessageListener {

    private static final String BROKER_URL="failover://tcp://127.0.0.1:61616";

    public static void main(String[] args) throws JMSException {

        //建立鏈接工廠
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(BROKER_URL);
        //得到鏈接
        Connection conn = connectionFactory.createConnection();
        //start
        conn.start();

        //建立Session，此方法第一個參數表示會話是否在事務中執行，第二個參數設定會話的應答模式
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //建立隊列
        Destination dest = session.createQueue("test-queue");
        //建立消息生產者
        MessageConsumer consumer = session.createConsumer(dest);

        //初始化MessageListener
        MsgConsumer msgConsumer = new MsgConsumer();

        //給消費者設定監聽對象
        consumer.setMessageListener(msgConsumer);

    }


    /**
     * 消費者須要實現MessageListener接口
     * 接口有一個onMessage(Message message)須要在此方法中作消息的處理
     */
    @Override
    public void onMessage(Message msg) {
        TextMessage txtMessage = (TextMessage)msg;
        try {
            System.out.println("get message:" + txtMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

依賴amq-core.jar    

compile 'org.apache.activemq:activemq-core:5.7.0'

先啓動amq， 瀏覽器輸入http://localhost:8161/ 進入amq控制檯。點擊Manage ActiveMQ broker，admin/admin 進入管理後臺

先運行生產者，關停，在運行消費者，觀察pending messages, messages enqueued,  message dequeued的數量變化。

注意：每次重啓amq，messages enqueued,  message dequeued會置0，pending messages表示待消費的消息(包括上次關閉前未消費的消息)。messages enqueued,  message dequeued表示本次開啓amq後的入隊(生產)、出隊(消費)消息。numbers of consumers表示消費者數量。開啓消費者時爲1， 關閉消費者時爲0