ActiveMQ簡介－互聯網 - 大數據

ActiveMQ是什么

　　ActiveMQ是Apache推出的一款開源的，完全支持JMS1.1和J2EE1.4規范的JMS Provider實現的消息中間件(Message Oriented Middleware,MOM)

ActiveMQ能干什么

　　最主要的功能是:實現JMS Provider,用來幫助實現高可用、高性能、可伸縮、易用和安全的企業級面向消息服務的系統

ActiveMQ的特點

　　完全支持JMS1.1和J2EE 1.4規范（持久化、XA消息、事務）
　　支持多種傳送協議: in-VM、TCP、SSL、NIO、UDP、JGroups、JXTA
　　可插拔的體系結構,可以靈活定制，如:消息存儲方式、安全管理等
　　很容易和Application Server集成使用
　　多種語言和協議編寫客戶端。如：Java、C、C++、PHP
　　可以很容易的和Spring結合使用

消息中間件
　　MOM基本功能: 將信息以消息的形式，從一個應用程序傳送到另一個或者多個應用程序
　　MOM的主要特點：
　　　　1.消息異步接受，類似于手機短信的行為，消息發送者不需要等待消息接受者的響應，減少軟件多系統集成的耦合度
　　　　2.消息可靠接收，確保消息在中間件可靠保存，只有接收方收到后才刪除消息，多個消息也可以組成原子事務
　　　　3.消息中間件的主要應用場景：
　　　　　　在多個系統間進行整合和通訊的時候，通常會要求：
　　　　　　　　a.可靠傳輸，數據不能丟失，有的時候，也會要求不能重復傳輸。
　　　　　　　　b.異步傳輸，否則各個系統同步發送接受數據，互相等待，造成系統瓶頸。

ActiveMQ安裝和基本使用
1.下載并安裝ActiveMQ服務器端，從http://activemq.apache.org/download.html下載最新的ActiveMQ,解壓后就可以直接使用
2.運行啟動
　　a.普通啟動，到ActiveMQ/bin目錄下面，執行 ./activemq start
　　b.啟動并指定日志文件: ./activemq start > /tmp/activemq.log
3.檢查是否已經啟動
　　ActiveMQ默認采用61616端口提供JMS服務，使用8161端口提供管理控制臺服務，執行以下命令以便檢查是否已經成功啟動ActiveMQ服務：
　　　　a. 比如查看61616端口是否打開： netstat -an | grep 61616
　　　　b. 直接訪問ActiveMQ的管理頁面： http://192.168.1.102:8161/admin/ 默認的用戶名和密碼是 admin/admin
4.關閉ActiveMQ, 可以用 ./activemq stop , 或者直接kill 線程

如果能啟動ActiveMQ后，訪問 http:// ip :8161/admin/ ,能夠出現下面的界面說明ActiveMQ就啟動成功了。

第一個ActiveMQ程序

1.新建一個Maven項目，在pom.xml里面添加ActiveMQ的lib依賴

 1       <dependency>
 2           <groupId>org.apache.activemq</groupId>
 3           <artifactId>activemq-all</artifactId>
 4           <version>5.9.0</version>
 5       </dependency>
 6   
 7     <dependency>
 8         <groupId>org.apache.xbean</groupId>
 9         <artifactId>xbean-spring</artifactId>
10         <version>3.16</version>
11     </dependency>

消息發送者:

 1 import javax.jms.Connection;
 2 import javax.jms.ConnectionFactory;
 3 import javax.jms.Destination;
 4 import javax.jms.MessageProducer;
 5 import javax.jms.Session;
 6 import javax.jms.TextMessage;
 7 
 8 import org.apache.activemq.ActiveMQConnectionFactory;
 9 
10 public class JmsSend {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.120:61616");
13         Connection connection = connectionFactory.createConnection();
14         connection.start();
15         
16         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17         Destination destination = session.createQueue("my-queue");
18         
19         MessageProducer producer = session.createProducer(destination);
20         for(int i = 0;i < 10;i++){
21             TextMessage message = session.createTextMessage("message --->" + i);
22             Thread.sleep(1000);
23             //通過生產者發出消息
24             producer.send(message);
25         }
26         session.commit();
27         session.close();
28         connection.close();
29     }
30 }

運行上面的消息發送者程序后，再去訪問：http:// ip :8161/admin/，會出現下面的界面：

消息接受者：

 1 import javax.jms.Connection;
 2 import javax.jms.ConnectionFactory;
 3 import javax.jms.Destination;
 4 import javax.jms.MessageConsumer;
 5 import javax.jms.Session;
 6 import javax.jms.TextMessage;
 7 
 8 import org.apache.activemq.ActiveMQConnectionFactory;
 9 
10 public class JmsReceiver {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://120.76.123.81:61616");
13         Connection connection =  cf.createConnection();
14         connection.start();
15         
16         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17         Destination destination = session.createQueue("my-queue");
18         MessageConsumer consumer = session.createConsumer(destination);
19         int i = 0;
20         while(i < 10){
21             i++;
22             TextMessage message = (TextMessage)consumer.receive();
23             session.commit();
24             System.out.println("接收到的消息是:"+message.getText());
25         }
26         session.close();
27         connection.close();
28     }
29 }