ActiveMQ介紹
ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的 JMS Provider實現，盡管JMS規范出台已經是很久的事情了，但是JMS在當今的J2EE應用中間仍然扮演着特殊的地位。
特性列表：
⒈ 多種語言和協議編寫客戶端。語言: Java,C,C++,C#,Ruby,Perl,Python,PHP。應用協議： OpenWire,Stomp REST,WS Notification,XMPP,AMQP
⒉ 完全支持JMS1.1和J2EE 1.4規范 （持久化，XA消息，事務)
⒊ 對spring的支持，ActiveMQ可以很容易內嵌到使用Spring的系統里面去，而且也支持Spring2.0的特性
⒋ 通過了常見J2EE服務器（如 Geronimo,JBoss 4,GlassFish,WebLogic)的測試，其中通過JCA 1.5 resource adaptors的配置，可以讓ActiveMQ可以自動的部署到任何兼容J2EE 1.4 商業服務器上
⒌ 支持多種傳送協議：in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
⒍ 支持通過JDBC和journal提供高速的消息持久化
⒎ 從設計上保證了高性能的集群，客戶端-服務器，點對點
⒏ 支持Ajax
⒐ 支持與Axis的整合
⒑ 可以很容易的調用內嵌JMS provider，進行測試
（二）ActiveMQ安裝、配置、啟動、可視化界面
1、安裝
下載地址：http://activemq.apache.org/download.html
2、配置（conf目錄下）
1）用戶名密碼設置 
2）開啟jmx監控
activemq.xml中進行如下修改 
注：這里的配置不是必須，根據需要自行配置
3、啟動
直接運行bin目錄下：activemq.bat
4、可視化界面
瀏覽器中：http://localhost:8161/admin/index.jsp
用戶名，密碼在：jetty-realm.properties中設置
（三）點對點式消息隊列（Queue）
消息生產者
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; public class QueueProducer { public static void main(String[] args) { //連接信息設置 String username = "system"; String password = "manager"; String brokerURL = "failover://tcp://localhost:61616"; //連接工廠 ConnectionFactory connectionFactory = null; //連接 Connection connection = null; //會話 接受或者發送消息的線程 Session session = null; //消息的目的地 Destination destination = null; //消息生產者 MessageProducer messageProducer = null; //實例化連接工廠 connectionFactory = new ActiveMQConnectionFactory(username, password, brokerURL); try { //通過連接工廠獲取連接 connection = connectionFactory.createConnection(); //啟動連接 connection.start(); //創建session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); //創建一個名稱為QueueTest的消息隊列 destination = session.createQueue("QueueTest"); //創建消息生產者 messageProducer = session.createProducer(destination); //發送消息 TextMessage message = null; for (int i=0; i<10; i++) { //創建要發送的文本信息 message = session.createTextMessage("Queue消息測試" +(i+1)); //通過消息生產者發出消息 messageProducer.send(message); System.out.println("發送成功：" + message.getText()); } session.commit(); } catch (Exception e) { e.printStackTrace(); }finally{ if(null != connection){ try { connection.close(); } catch (JMSException e) { e.printStackTrace(); } } } } }
消息消費者
import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; public class QueueConsumer { public static void main(String[] args) { //連接信息設置 String username = "system"; String password = "manager"; String brokerURL = "failover://tcp://localhost:61616"; //連接工廠 ConnectionFactory connectionFactory = null; //連接 Connection connection = null; //會話 接受或者發送消息的線程 Session session = null; //消息的目的地 Destination destination = null; //消息消費者 MessageConsumer messageConsumer = null; //實例化連接工廠 connectionFactory = new ActiveMQConnectionFactory(username, password, brokerURL); try { //通過連接工廠獲取連接 connection = connectionFactory.createConnection(); //啟動連接 connection.start(); //創建session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //創建一個連接QueueTest的消息隊列 destination = session.createQueue("QueueTest"); //創建消息消費者 messageConsumer = session.createConsumer(destination); while (true) { TextMessage textMessage = (TextMessage) messageConsumer.receive(100000); if(textMessage != null){ System.out.println("成功接收消息:" + textMessage.getText()); }else { break; } } } catch (JMSException e) { e.printStackTrace(); } }
（四）主題發布訂閱式（Topic）
主題發布者
import javax.jms.Connection; import javax.jms.ConnectionFactory; 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.ActiveMQConnectionFactory; public class TopicProducer { public static void main(String[] args) { //連接信息設置 String username = "system"; String password = "manager"; String brokerURL = "failover://tcp://localhost:61616"; //連接工廠 ConnectionFactory connectionFactory = null; //連接 Connection connection = null; //會話 接受或者發送消息的線程 Session session = null; //消息的主題 Topic topic = null; //消息生產者 MessageProducer messageProducer = null; //實例化連接工廠 connectionFactory = new ActiveMQConnectionFactory(username, password, brokerURL); try { //通過連接工廠獲取連接 connection = connectionFactory.createConnection(); //啟動連接 connection.start(); //創建session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); //創建名為TopicTest的主題 topic = session.createTopic("TopicTest"); //創建主題生產者 messageProducer = session.createProducer(topic); messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);//不將數據持久化 //發送主題 TextMessage message = null; for (int i=0; i<10; i++) { //創建要發送的文本信息 message = session.createTextMessage("Topic主題測試" +(i+1)); //通過主題生產者發出消息 messageProducer.send(message); System.out.println("發送成功：" + message.getText()); } session.commit(); } catch (Exception e) { e.printStackTrace(); }finally{ if(null != connection){ try { connection.close(); } catch (JMSException e) { e.printStackTrace(); } } } } }
主題訂閱者
import javax.jms.Connection; import javax.jms.ConnectionFactory; 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.ActiveMQConnectionFactory; public class TopicConsumer { public static void main(String[] args) { //連接信息設置 String username = "system"; String password = "manager"; String brokerURL = "failover://tcp://localhost:61616"; //連接工廠 ConnectionFactory connectionFactory = null; //連接 Connection connection = null; //會話 接受或者發送消息的線程 Session session = null; //主題的目的地 Topic topic = null; //主題消費者 MessageConsumer messageConsumer = null; //實例化連接工廠 connectionFactory = new ActiveMQConnectionFactory(username, password, brokerURL); try { //通過連接工廠獲取連接 connection = connectionFactory.createConnection(); //啟動連接 connection.start(); //創建session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //創建一個連接TopicTest的主題 topic = session.createTopic("TopicTest"); //創建主題消費者 messageConsumer = session.createConsumer(topic); messageConsumer.setMessageListener(new MyMessageListener()); } catch (JMSException e) { e.printStackTrace(); } } } class MyMessageListener implements MessageListener { @Override public void onMessage(Message message) { TextMessage textMessage = (TextMessage) message; try { System.out.println("接收訂閱主題：" + textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } }
注：
1、代碼中所需額外jar包在下載的mq文件夾中，例如我使用的：activemq-all-5.9.0.jar
2、對於消息隊列，異步生產和消費；對於主題發布訂閱要先啟動訂閱者進行監聽，然后在發布方可接收到訂閱主題
3、關於Queue與Topic的具體區別，詳見http://blog.csdn.net/qq_21033663/article/details/52458305