1.?ActiveMQ

1). ActiveMQ

ActiveMQ是Apache所提供的一個開源的消息系統，完全采用Java來實現，因此，它能很好地支持J2EE提出的JMS（Java Message Service,即Java消息服務）規范。JMS是一組Java應用程序接口，它提供消息的創建、發送、讀取等一系列服務。JMS提供了一組公共應用程序接口和響應的語法，類似于Java數據庫的統一訪問接口JDBC,它是一種與廠商無關的API，使得Java程序能夠與不同廠商的消息組件很好地進行通信。

2). Java Message Service(JMS)

JMS支持兩種消息發送和接收模型。

• 一種稱為P2P(Ponit to Point)模型，即采用點對點的方式發送消息。P2P模型是基于隊列的，消息生產者發送消息到隊列，消息消費者從隊列中接收消息，隊列的存在使得消息的異步傳輸稱為可能，P2P模型在點對點的情況下進行消息傳遞時采用。

  ?

  圖1.png

• 另一種稱為Pub/Sub(Publish/Subscribe，即發布-訂閱)模型，發布-訂閱模型定義了如何向一個內容節點發布和訂閱消息，這個內容節點稱為topic(主題)。主題可以認為是消息傳遞的中介，消息發布這將消息發布到某個主題，而消息訂閱者則從主題訂閱消息。主題使得消息的訂閱者與消息的發布者互相保持獨立，不需要進行接觸即可保證消息的傳遞，發布-訂閱模型在消息的一對多廣播時采用。

  ?

  圖2.png

3). JMS術語

• Provider/MessageProvider：生產者
• Consumer/MessageConsumer：消費者
• PTP：Point To Point，點對點通信消息模型
• Pub/Sub：Publish/Subscribe，發布訂閱消息模型
• Queue：隊列，目標類型之一，和PTP結合
• Topic：主題，目標類型之一，和Pub/Sub結合
• ConnectionFactory：連接工廠，JMS用它創建連接
• Connnection：JMS Client到JMS Provider的連接
• Destination：消息目的地，由Session創建
• Session：會話，由Connection創建，實質上就是發送、接受消息的一個線程，因此生產者、消費者都是Session創建的

4).?ActiveMQ下載

圖3.png

• bin (windows下面的bat(分32、64位)和unix/linux下面的sh)
• conf (activeMQ配置目錄，包含最基本的activeMQ配置文件）
• data (默認是空的)
• docs (index,replease版本里面沒有文檔，-.-b不知道為啥不帶）
• example (幾個例子)
• lib (activemMQ使用到的lib)
• webapps 注意ActiveMQ自帶Jetty提供Web管控臺
• webapps-demo 示例
• activemq-all-5.15.3.jar
• LICENSE.txt
• README.txt

5). 配置

• Web控制臺賬號和密碼(apache-activemq-5.15.3\conf)

  圖4.png

• 網絡端口(apache-activemq-5.15.3\conf)--默認為8161

  圖5.png

6). 啟動

\apache-activemq-5.15.3\bin\win64\目錄下雙擊activemq.bat文件，在瀏覽器中輸入http://localhost:8161/admin/, 用戶名和密碼輸入admin即可

圖6.png

?

7). 消息中間件(MOM：Message Orient middleware)

消息中間件有很多的用途和優點：

• 1 將數據從一個應用程序傳送到另一個應用程序，或者從軟件的一個模塊傳送到另外一個模塊；
• 負責建立網絡通信的通道，進行數據的可靠傳送。
• 保證數據不重發，不丟失
• 能夠實現跨平臺操作，能夠為不同操作系統上的軟件集成技工數據傳送服務

8).什么情況下使用ActiveMQ?

• 多個項目之間集成
  (1) 跨平臺
  (2) 多語言
  (3) 多項目
• 降低系統間模塊的耦合度，解耦
  (1) 軟件擴展性
• 系統前后端隔離
  (1) 前后端隔離，屏蔽高安全區

2. ActiveMQ 示例

1). P2P 示例

I. 導包--activemq-all-5.15.3.jar
II. Producer

/*** 定義消息的生產者* @author mazaiting*/ public class Producer {// 用戶名private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;// 密碼private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;// 鏈接private static final String BROKENURL = ActiveMQConnection.DEFAULT_BROKER_URL;/*** 定義消息并發送，等待消息的接收者(消費者)消費此消息* @param args* @throws JMSException */public static void main(String[] args) throws JMSException {// 消息中間件的鏈接工廠ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKENURL);// 連接Connection connection = null;// 會話Session session = null;// 消息的目的地Destination destination = null;// 消息生產者MessageProducer messageProducer = null;try {// 通過連接工廠獲取鏈接connection = connectionFactory.createConnection();// 創建會話，進行消息的發送// 參數一：是否啟用事務// 參數二：設置自動簽收session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);// 創建消息隊列destination = session.createQueue("talkWithMo");// 創建一個消息生產者messageProducer = session.createProducer(destination);// 設置持久化/非持久化， 如果非持久化，MQ重啟后可能后導致消息丟失messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 模擬發送消息for (int i = 0; i < 5; i++) {TextMessage textMessage = session.createTextMessage("給媽媽發送的消息:"+i);System.out.println("textMessage: " + textMessage);messageProducer.send(textMessage);}// 如果設置了事務，會話就必須提交session.commit();} catch (JMSException e) {e.printStackTrace();} finally {if (null != connection) {connection.close();}}} }

III. Consumer

/*** 定義消息的消費者* @author mazaiting*/ public class Consumer {// 用戶名private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;// 密碼private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;// 鏈接private static final String BROKENURL = ActiveMQConnection.DEFAULT_BROKER_URL;/*** 接收消息* @param args* @throws JMSException */public static void main(String[] args) throws JMSException {// 消息中間件的鏈接工廠ConnectionFactory connectionFactory = null;// 鏈接Connection connection = null;// 會話Session session = null;// 消息的目的地Destination destination = null;// 消息的消費者MessageConsumer messageConsumer = null;// 實例化鏈接工廠，創建一個鏈接connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKENURL);try {// 通過工廠獲取鏈接connection = connectionFactory.createConnection();// 啟動鏈接connection.start();// 創建會話，進行消息的接收session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);// 創建消息隊列destination = session.createQueue("talkWithMo");// 創建一個消息的消費者messageConsumer = session.createConsumer(destination);// 模擬接收消息while (true) {TextMessage textMessage = (TextMessage) messageConsumer.receive(10000);if (null != textMessage) {System.out.println("收到消息： " + textMessage);} else {break;}}// 提交session.commit();} catch (JMSException e) {e.printStackTrace();} finally {if (null != connection) {connection.close();}}} }

IV. 測試

• 先運行生產者Producer

  ?

  圖7.png

ActiveMQ控制臺

?

圖8.png

• 再運行消費者Consumer

  ?

  圖9.png

  ?

  ActiveMQ控制臺

  ?

  圖10.png

V. 消息類型

• StreamMessage Java原始值的數據流
• MapMessage 一套名稱-鍵值對
• TextMessage 一個字符串對象
• ObjectMessage 一個序列號的Java對象
• BytesMessage 一個未解釋字節的數據流
  VI. 控制臺 Queue
• Messages Enqueued：表示生產了多少條消息，記做P
• Messages Dequeued：表示消費了多少條消息，記做C
• Number Of Consumers：表示在該隊列上還有多少消費者在等待接受消息
• Number Of Pending Messages：表示還有多少條消息沒有被消費，實際上是表示消息的積壓程度，就是P-C
  VII. 簽收
  簽收就是消費者接受到消息后，需要告訴消息服務器，我收到消息了。當消息服務器收到回執后，本條消息將失效。因此簽收將對PTP模式產生很大影響。如果消費者收到消息后，并不簽收，那么本條消息繼續有效，很可能會被其他消費者消費掉！
• AUTO_ACKNOWLEDGE：表示在消費者receive消息的時候自動的簽收
• CLIENT_ACKNOWLEDGE：表示消費者receive消息后必須手動的調用acknowledge()方法進行簽收
• DUPS_OK_ACKNOWLEDGE：簽不簽收無所謂了，只要消費者能夠容忍重復的消息接受，當然這樣會降低Session的開銷

2). request/reply模型

I. 實現思路

?

圖11.png

?

Client的Producer發出一個JMS message形式的request，request上附加了一些額外的屬性：

• correlation ID(用來和返回的correlation ID對比進行驗證)，
• JMSReplyTo屬性(放置jms message的destination，這樣worker的Consumer獲得jms message就能得到destination)

Worker的consumer收到requset，處理request并用producer發出reply，destination就從requset的JMSReplyTo屬性中得到。

II. Server代碼

public class Server implements MessageListener {// 經紀人鏈接private static final String BROKER_URL = "tcp://localhost:61616";// 請求隊列private static final String REQUEST_QUEUE = "requestQueue";// 經紀人服務private BrokerService brokerService;// 會話private Session session;// 生產者private MessageProducer producer;// 消費者private MessageConsumer consumer;private void start() throws Exception {createBroker();setUpConsumer();}/*** 創建經紀人* @throws Exception */private void createBroker() throws Exception {// 創建經紀人服務brokerService = new BrokerService();// 設置是否持久化brokerService.setPersistent(false);// 設置是否使用JMXbrokerService.setUseJmx(false);// 添加鏈接brokerService.addConnector(BROKER_URL);// 啟動brokerService.start();}/*** 設置消費者* @throws JMSException */private void setUpConsumer() throws JMSException {// 創建連接工廠ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);// 創建連接Connection connection = connectionFactory.createConnection();// 啟動連接connection.start();// 創建Sessionsession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 創建隊列Destination adminQueue = session.createQueue(REQUEST_QUEUE);// 創建生產者producer = session.createProducer(null);// 設置持久化模式producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 創建消費者consumer = session.createConsumer(adminQueue);// 消費者設置消息監聽consumer.setMessageListener(this);}public void stop() throws Exception {producer.close();consumer.close();session.close();brokerService.stop();}@Overridepublic void onMessage(Message message) {try {// 創建新消息TextMessage response = this.session.createTextMessage();// 判斷消息是否是文本消息if (message instanceof TextMessage) {// 強轉為文本消息 TextMessage textMessage = (TextMessage) message;// 獲取消息內容String text = textMessage.getText();// 設置消息response.setText(handleRequest(text));}response.setJMSCorrelationID(message.getJMSCorrelationID());producer.send(message.getJMSReplyTo(), response);} catch (JMSException e) {e.printStackTrace();}}/*** 構建消息內容* @param text 文本* @return*/private String handleRequest(String text) {return "Response to '" + text + "'";}public static void main(String[] args) throws Exception {Server server = new Server();// 啟動server.start();System.out.println();System.out.println("Press any key to stop the server");System.out.println();System.in.read();server.stop();} }

III. Client代碼

public class Client implements MessageListener {// 經紀人鏈接private static final String BROKER_URL = "tcp://localhost:61616";// 請求隊列private static final String REQUEST_QUEUE = "requestQueue";// 連接private Connection connection;// 會話private Session session;// 生產者private MessageProducer producer;// 消費者private MessageConsumer consumer;// 請求隊列private Queue tempDest;public void start() throws JMSException {// 連接工廠ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(BROKER_URL);// 創建連接connection = activeMQConnectionFactory.createConnection();// 開啟連接connection.start();// 創建會話session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 創建隊列Destination adminQueue = session.createQueue(REQUEST_QUEUE);// 創建生產者producer = session.createProducer(adminQueue);// 設置持久化模式producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 創建模板隊列tempDest = session.createTemporaryQueue();// 創建消費者consumer = session.createConsumer(tempDest);// 設置消息監聽consumer.setMessageListener(this); }/*** 停止* @throws JMSException */public void stop() throws JMSException {producer.close();consumer.close();session.close();}/*** 請求* @param request* @throws JMSException */public void request(String request) throws JMSException {System.out.println("Request: " + request);// 創建文本消息TextMessage textMessage = session.createTextMessage();// 設置文本內容textMessage.setText(request);// 設置回復textMessage.setJMSReplyTo(tempDest);// 獲取UUIDString correlationId = UUID.randomUUID().toString();// 設置JMS idtextMessage.setJMSCorrelationID(correlationId);// 發送消息this.producer.send(textMessage);}@Overridepublic void onMessage(Message message) {try {System.out.println("Received response for: " + ((TextMessage)message).getText());} catch (JMSException e) {e.printStackTrace();}}public static void main(String[] args) throws JMSException, InterruptedException {Client client = new Client();// 啟動client.start();int i = 0;while(i++ < 10) {client.request("REQUEST- " + i);}Thread.sleep(3000);client.stop();} }

IV. 測試

• 啟動Server

  ?

  圖12.png

• 啟動Client

  ?

  圖13.png

總結

以上是生活随笔為你收集整理的ActiveMQ 消息中间件的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。