ActiveMQ 是Apache出品，最流行的，能力強勁的開源訊息匯流排。ActiveMQ 是一個完全支援JMS1.1（Java訊息服務）和J2EE 1.4規範的 JMS Provider實現。

1、ActiveMQ下載

有windows和linux兩個版本的，分為32和64位，具體按需求下載
2、解壓、測試連線
將下載的安裝包【apache-activemq-5.8.0-bin.zip】解壓，有如下目錄：

進入bin目錄，雙擊執行【activemq.bat】檔案，出現如下介面，說明服務已啟動

通過點選選單欄Queues進入佇列管理介面，可建立一個佇列

（1）Number Of Pending Messages 等待消費的訊息 這個是當前未出佇列的數量。可以理解為總接收數-總出佇列數
（2）Number Of Consumers 消費者 這個是消費者端的消費者數量
（3）Messages Enqueued 進入佇列的訊息 進入佇列的總數量,包括出佇列的。 這個數量只增不減
（4）Messages Dequeued 出了佇列的訊息 可以理解為是消費這消費掉的數量

3、在eclipse中配置activeMq
建立Java project

所需mq jar包在解壓目錄的lib中匯入專案即可，建立兩java類，一個訊息傳送方，一個接受方。
【訊息傳送發】：

import java.util.Random;        
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
importorg.apache.activemq.ActiveMQConnectionFactory;

public class SendMessage {  
    private static final String url = "tcp://localhost:61616";  
    private static final String QUEUE_NAME = "bfp_mq";  

    public void sendMessage() throws JMSException {  
   // JMS 客戶端到JMSProvider 的連線  
   Connection connection = null;  
   try {  
       // 連線工廠，JMS 用它建立連線  
       // 構造ConnectionFactory例項物件，此處採用ActiveMq的實現jar  
       ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);  
       connection = (Connection)connectionFactory.createConnection();  
       // 啟動連線  
       connection.start();  
       //Session：傳送或接收訊息的執行緒  
       // 獲取session  
       Session session = (Session) connection.createSession(false,  
              Session.AUTO_ACKNOWLEDGE);  
       // 訊息的目的地，訊息傳送到那個佇列  
       Destination destination = session.createQueue(QUEUE_NAME);  
       //MessageProducer：訊息傳送者（生產者）  
       // 建立訊息傳送者  
       MessageProducer producer =session.createProducer(destination);  
       // 設定是否持久化  
       //DeliveryMode.NON_PERSISTENT：不持久化  
       //DeliveryMode.PERSISTENT：持久化  
       producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);  

       String msg = "";  
       int i = 0;  
    do {  
        msg = "第"+i + "次傳送的訊息："+new Random();  
            TextMessage message = session.createTextMessage(msg);  
            Thread.sleep(1000);  
            // 傳送訊息到目的地方  
           producer.send(message);  
            System.out.println("傳送訊息：" +msg);  
            i++;  
    } while (i<1000);  
   } catch (Exception e) {  
       e.printStackTrace();  
   }  
}  

public static void main(String[] args) {  
   SendMessage sndMsg = new SendMessage();  
   try {  
       sndMsg.sendMessage();  
   } catch (Exception ex) {  
       System.out.println(ex.toString());  
   }  
}  
 

}

【訊息消費方】：

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.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/** 
 * 訊息接收類 
 * 
 * @createTime:Apr 7, 2013 5:11:11 PM 
 * @author:<a href="mailto:
 
[email protected]">迷蝶</a> 
 * @version:0.1 
 *@lastVersion: 0.1 
 * @updateTime: 
 *@updateAuthor: <a href="mailto:[email protected]">迷蝶</a> 
 * @changesSum: 
 * 
 */  
public class ReceiveMessage {  
         private static final String url = "tcp://localhost:61616";  
         private static final String QUEUE_NAME = "bfp_mq";  

     public void receiveMessage() {  
               Connection connection = null;  
               try{  
                        try{  
                                 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(  
                                                    url);  
                                 connection= connectionFactory.createConnection();  
                        }catch (Exception e) {  
                                 System.out.println(e.toString());  
                        }  
                        connection.start();  
                        Session session = connection.createSession(false,  
                                           Session.AUTO_ACKNOWLEDGE);  
                        Destination destination = session.createQueue(QUEUE_NAME);  
                        //訊息接收者，也就是消費者  
                        MessageConsumer consumer = session.createConsumer(destination);  

                        consumeMessagesAndClose(connection,session, consumer);  
               }catch (Exception e) {  
                        System.out.println(e.toString());  
               }  
     }  
     /** 
      * 接收和關閉訊息，如遇到訊息內容為close則，關閉連線 
      * 
      * @param connection   JMS 客戶端到JMSProvider 的連線 
      * @param session                   傳送或接收訊息的執行緒 
      * @param consumer              訊息接收物件 
      * @throws JMSException 
      * @auther <ahref="mailto:[email protected]">迷蝶</a> 
      * Apr 8, 2013 10:31:55 AM 
      */  
     protected void consumeMessagesAndClose(Connection connection,  
                        Session session, MessageConsumer consumer) throws JMSException {  
               do{  
                        Message message = consumer.receive(1000);  
                        if("close".equals(message)){  
                                 consumer.close();  
                                 session.close();  
                                 connection.close();  
                        }  
                        if(message != null) {  
                                 onMessage(message);  
                        }  
               }while (true);  

     }  

     public void onMessage(Message message) {  
               try{  
                        if(message instanceof TextMessage) {  
                                 TextMessage txtMsg = (TextMessage) message;  
                                 String msg = txtMsg.getText();  
                                 System.out.println("Received:" + msg);  
                        }  
               }catch (Exception e) {  
                        e.printStackTrace();  
               }  

     }  

     public static void main(String args[]) {  
               ReceiveMessage rm = new ReceiveMessage();  
               rm.receiveMessage();  
     }  

}

可執行程式碼，在訊息管理介面檢視待發送、進入佇列、出佇列的訊息以及消費者個數。