ActiveMQ的安裝、運用

1、下載ActiveMQ：http://activemq.apache.org/

2、解壓包，并找到activemq.bat,雙擊打開一個控制臺，跟你打開tomcat一樣的。如圖：

3、在瀏覽器查看你的activemq是否正確啟動：http://localhost:8161/admin,打開之后看到這樣：  

用戶名：admin  密碼：admin

確定即可登錄。

4、你可以新建一個Queue，在  Queue Name后面的方框中填入“MyFirstQueue”，點擊正后方“Create”按鈕即可。

在下面Queues：列表中，就會顯示你剛剛新建的Queue：

這是已經使用過的，其實一開始數值是這樣的：   

Name    ：      MyFirstQueue     

Number Of   ending Message  ： 0

Number Of   Consumers ：0

Message Enqueued：0

Message Dequeued：0

5、接下來我們通過IDEA來創建兩個java類，一個是消息生產者，一個是消息消費者。

這是消息生產者：

package com.zfm.activemq;

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

import javax.jms.*;

public class Sender {
    private static final int SEND_NUMBER =5;
    public static void main(String[] args){
        //ConnectionFactory:連接工廠
        ConnectionFactory  connectionFactory;
        //Connection:JMS客戶端到JMS Provider的連接
        Connection connection = null;
        //Session :一個會話，發送或接收消息的線程
        Session session;
        //Destination:消息的目的地
        Destination destination;
        //MessageProducer：消息產生者：發送
        MessageProducer messageProducer;
        //構造connectionFactory實例對象

        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://localhost:61616"/*ActiveMQ默認使用的TCP連接端口是61616,*/
        );

        try{
            //構造從工廠得到的連接
            connection = connectionFactory.createConnection();
            //啟動
            connection.start();
            //獲取操作連接
            session = connection.createSession(Boolean.TRUE,
                    Session.AUTO_ACKNOWLEDGE);
            //獲取session的參數

            destination = session.createQueue("FMDemo");
            //得到消息生成者
            messageProducer = session.createProducer(destination);

            //設置·不持久化
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            //構造消息
            sendMeaasge(session,messageProducer);
            session.commit();


        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try{
                if(null!=connection){
                    connection.close();
                }
            }catch (Throwable t){

            }
        }
    }

    public static void sendMeaasge(Session session,MessageProducer producer) throws JMSException {
        for(int i=1;i<=SEND_NUMBER;i++){
            TextMessage  message = session.createTextMessage("ActiveMq 發送的消息" +i);
            //發送消息到目的地
            System.out.println("發送消息： ActiveMq發送消息：" + i);
            producer.send(message);
        }

    }
}

消息消費者：

package com.zfm.activemq;

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

import javax.jms.*;

public class Receiver {
    public static void main(String[] rags) throws JMSException {
        //Connection連接工廠
        ConnectionFactory connectionFactory;
        //Connection:JMS客戶端到JMS provider
        Connection connection = null;
        //Session : 一個發送或接受消息的·會話
        Session session;
        //Destination:消息的目的地,消息發送給誰
        Destination destination;
        //消費者，消息接收者
        MessageConsumer messageConsumer;

        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://localhost:61616"http://ActiveMQ默認使用的TCP連接端口是61616,
        );
        try{
            //從構造工廠里獲得連接
            connection = connectionFactory.createConnection();
            //一定要啟動
            connection.start();

            //從連接獲得會話
            session = connection.createSession(Boolean.FALSE
            ,Session.AUTO_ACKNOWLEDGE);

            destination = session.createQueue("FMDemo");
            messageConsumer = session.createConsumer(destination);
            while(true){
                //設置接收者接受新消息的時間
                TextMessage message = (TextMessage)messageConsumer.receive(100000);
                if(null!=message){
                    System.out.println("收到消息" + message.getText());
                }else{
                    break;
                }
            }



        }catch (Exception e){

        }finally {
            try{
                if(null != connection) {
                    connection.close();
                }
            }catch(Throwable t){

            }
        }

    }
}

當運行Sender.java之后，刷新http://localhost:8161/admin/queues.jsp時，

Queues表中的參數有變化咯：

Name    ：     MyFirstQueue

Number Of   Pending Message  ：5

Number Of   Consumers ：0

Message  Enqueued ：5

Message  Dequeued    ：0

此時，由于消息已經發完，并且我們沒有讓Sender的main函數一直運行，所以，在控制臺打印了：

發送消息： ActiveMq發送消息：1
發送消息： ActiveMq發送消息：2
發送消息： ActiveMq發送消息：3
發送消息： ActiveMq發送消息：4
發送消息： ActiveMq發送消息：5

Process finished with exit code 0

進程結束了。

這時我們再運行Receiver.java，而后控制臺打印：

收到消息ActiveMq 發送的消息1
收到消息ActiveMq 發送的消息2
收到消息ActiveMq 發送的消息3
收到消息ActiveMq 發送的消息4
收到消息ActiveMq 發送的消息5

Process finished with exit code 0

此時，再刷新http://localhost:8161/admin/queues.jsp，此時表格數值變為：

Name    ：     MyFirstQueue

Number Of   Pending Message  ：0

Number Of   Consumers ：0

Message  Enqueued ：5

Message  Dequeued    ：5

還有一點是：當Receiver.java并沒有運行結束的時候，由于我只開啟了一個Receiver進程，所以這時Number Of   Consumers：1。

可見，由Sender發出的消息已經被Receiver收到了。你可以先把Sender消息發出來，這個時候只要消息已經到消息隊列上了，只要你不期望發送者還要接收什么回復，你就可以把Sender停掉了。Receiver還是一樣的接收到信息。

下面介紹一下ActiveMQ的幾種基本通信方式：發布-訂閱模式和點對點模式。

基礎流程：

1、獲得ActiveMQConnectionFactory。

2、利用factory獲得Connection。

3、啟動connection。

4、通過connection創建Session。

5、指定Session的Destination。

6、發送消息者則創建MessageProducer/接收消息者創建MessageConsumer。

7、發送和接收JMS Message。

8、關閉所有JMS資源。

ActiveMQ詳情見http://shmilyaw-hotmail-com.iteye.com/blog/1897635

講到ActiveMQ，其實他是java消息服務（JMS）的其中一種規范，JMS詳情見：http://blog.csdn.net/jiuqiyuliang/article/details/46701559   和   http://blog.csdn.net/jiuqiyuliang/article/details/47160259。

什么情況下使用ActiveMQ?

1. 多個項目之間集成
   (1) 跨平臺
   (2) 多語言
   (3) 多項目

2. 降低系統間模塊的耦合度，解耦
   (1) 軟件擴展性

3. 系統前后端隔離
   (1) 前后端隔離，屏蔽高安全區