ActiveMQ消息隊列的使用
一、ActiveMQ簡介
ActiveMQ是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規範的 JMS Provider實現,盡管JMS規範出臺已經是很久的事情了,但是JMS在當今的J2EE應用中間仍然扮演著特殊的地位。
主要特點：
1. 多種語言和協議編寫客戶端。語言: Java, C, C++, C#, Ruby, Perl, Python, PHP。應用協議: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
2. 完全支持JMS1.1和J2EE 1.4規範 (持久化,XA消息,事務)
3. 對Spring的支持,ActiveMQ可以很容易內嵌到使用Spring的系統裏面去,而且也支持Spring2.0的特性
4. 通過了常見J2EE服務器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的測試,其中通過JCA 1.5 resource adaptors的配置,可以讓ActiveMQ可以自動的部署到任何兼容J2EE 1.4 商業服務器上
5. 支持多種傳送協議:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
6. 支持通過JDBC和journal提供高速的消息持久化
7. 從設計上保證了高性能的集群,客戶端-服務器,點對點
8. 支持Ajax
9. 支持與Axis的整合
10. 可以很容易得調用內嵌JMS provider,進行測試
二、ActiveMQ的消息形式
對於消息的傳遞有兩種類型：
一種是點對點的（Queue），即一個生產者和一個消費者一一對應；
另一種是發布/訂閱模式（Topic），即一個生產者產生消息並進行發送後，可以由多個消費者進行接收。
JMS定義了五種不同的消息正文格式，以及調用的消息類型，允許你發送並接收以一些不同形式的數據，提供現有消息格式的一些級別的兼容性。
- StreamMessage -- Java原始值的數據流
- MapMessage--一套名稱-值對
- TextMessage--一個字符串對象
- ObjectMessage--一個序列化的 Java對象
- BytesMessage--一個字節的數據流
三、ActiveMQ的安裝
第一步： 把ActiveMQ 的壓縮包上傳到Linux系統。
第二步：解壓縮。
第三步：啟動。
使用bin目錄下的activemq命令啟動：
[root@localhost bin]# ./activemq start
關閉：
[root@localhost bin]# ./activemq stop
查看狀態：
[root@localhost bin]# ./activemq status
註意：如果ActiveMQ整合spring使用不要使用activemq-all-5.12.0.jar包。建議使用5.11.2
進入管理後臺：
http://192.168.25.168:8161/admin
用戶名：admin
密碼：admin
四、ActiveMQ的兩種使用方法
1、點對點方式（如果沒有人處理，就會持久化儲存知道有人來處理）
public class ActiveMqTest {
/**
* 點到點形式發送消息
*/
@Test
public void testQueueProducer() throws Exception {
//1、創建一個連接工廠對象，需要指定服務的ip及端口。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.161:61616");
//2、使用工廠對象創建一個Connection對象。
Connection connection = connectionFactory.createConnection();
//3、開啟連接，調用Connection對象的start方法。
connection.start();
//4、創建一個Session對象。
//第一個參數：是否開啟事務。如果true開啟事務，第二個參數無意義。一般不開啟事務false。
//第二個參數：應答模式。自動應答或者手動應答。一般自動應答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、使用Session對象創建一個Destination對象。兩種形式queue、topic，現在應該使用queue
Queue queue = session.createQueue("test-queue");
//6、使用Session對象創建一個Producer對象。
MessageProducer producer = session.createProducer(queue);
//7、創建一個Message對象，可以使用TextMessage。
/*TextMessage textMessage = new ActiveMQTextMessage();
textMessage.setText("hello Activemq");*/
TextMessage textMessage = session.createTextMessage("hello activemq");
//8、發送消息
producer.send(textMessage);
//9、關閉資源
producer.close();
session.close();
connection.close();
}
@Test
public void testQueueConsumer() throws Exception {
//創建一個ConnectionFactory對象連接MQ服務器
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.161:61616");
//創建一個連接對象
Connection connection = connectionFactory.createConnection();
//開啟連接
connection.start();
//使用Connection對象創建一個Session對象
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//創建一個Destination對象。queue對象
Queue queue = session.createQueue("test-queue");
//使用Session對象創建一個消費者對象。
MessageConsumer consumer = session.createConsumer(queue);
//接收消息
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
//打印結果
TextMessage textMessage = (TextMessage) message;
String text;
try {
text = textMessage.getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//等待接收消息
System.in.read();
//關閉資源
consumer.close();
session.close();
connection.close();
}
}
2、廣播方式（只會廣播在線的，過時不候）
public class ActiveMqTest {
@Test
public void testTopicProducer() throws Exception {
//1、創建一個連接工廠對象，需要指定服務的ip及端口。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.161:61616");
//2、使用工廠對象創建一個Connection對象。
Connection connection = connectionFactory.createConnection();
//3、開啟連接，調用Connection對象的start方法。
connection.start();
//4、創建一個Session對象。
//第一個參數：是否開啟事務。如果true開啟事務，第二個參數無意義。一般不開啟事務false。
//第二個參數：應答模式。自動應答或者手動應答。一般自動應答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、使用Session對象創建一個Destination對象。兩種形式queue、topic，現在應該使用topic
Topic topic = session.createTopic("test-topic");
//6、使用Session對象創建一個Producer對象。
MessageProducer producer = session.createProducer(topic);
//7、創建一個Message對象，可以使用TextMessage。
/*TextMessage textMessage = new ActiveMQTextMessage();
textMessage.setText("hello Activemq");*/
TextMessage textMessage = session.createTextMessage("topic message");
//8、發送消息
producer.send(textMessage);
//9、關閉資源
producer.close();
session.close();
connection.close();
}
@Test
public void testTopicConsumer() throws Exception {
//創建一個ConnectionFactory對象連接MQ服務器
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.161:61616");
//創建一個連接對象
Connection connection = connectionFactory.createConnection();
//開啟連接
connection.start();
//使用Connection對象創建一個Session對象
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//創建一個Destination對象。topic對象
Topic topic = session.createTopic("test-topic");
//使用Session對象創建一個消費者對象。
MessageConsumer consumer = session.createConsumer(topic);
//接收消息
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
//打印結果
TextMessage textMessage = (TextMessage) message;
String text;
try {
text = textMessage.getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
System.out.println("topic消費者3啟動。。。。");
//等待接收消息
System.in.read();
//關閉資源
consumer.close();
session.close();
connection.close();
}
}
ActiveMQ消息隊列的使用