1. ActiveMQ

1). ActiveMQ

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

2). Java Message Service(JMS)

JMS支持兩種消息發(fā)送和接收模型。

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

  
  圖1.png

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

  
  圖2.png

3). JMS術(shù)語

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

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
• 網(wǎng)絡(luò)端口(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)

消息中間件有很多的用途和優(yōu)點：

• 1 將數(shù)據(jù)從一個應(yīng)用程序傳送到另一個應(yīng)用程序，或者從軟件的一個模塊傳送到另外一個模塊；
• 負責(zé)建立網(wǎng)絡(luò)通信的通道，進行數(shù)據(jù)的可靠傳送。
• 保證數(shù)據(jù)不重發(fā)，不丟失
• 能夠?qū)崿F(xiàn)跨平臺操作，能夠為不同操作系統(tǒng)上的軟件集成技工數(shù)據(jù)傳送服務(wù)

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

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

2. ActiveMQ 示例

1). P2P 示例

I. 導(dǎo)包--activemq-all-5.15.3.jar
II. Producer

/*** 定義消息的生產(chǎn)者* @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; /** * 定義消息并發(fā)送，等待消息的接收者(消費者)消費此消息 * @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; // 消息生產(chǎn)者 MessageProducer messageProducer = null; try { // 通過連接工廠獲取鏈接 connection = connectionFactory.createConnection(); // 創(chuàng)建會話，進行消息的發(fā)送 // 參數(shù)一：是否啟用事務(wù) // 參數(shù)二：設(shè)置自動簽收 session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); // 創(chuàng)建消息隊列 destination = session.createQueue("talkWithMo"); // 創(chuàng)建一個消息生產(chǎn)者 messageProducer = session.createProducer(destination); // 設(shè)置持久化/非持久化， 如果非持久化，MQ重啟后可能后導(dǎo)致消息丟失 messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 模擬發(fā)送消息 for (int i = 0; i < 5; i++) { TextMessage textMessage = session.createTextMessage("給媽媽發(fā)送的消息:"+i); System.out.println("textMessage: " + textMessage); messageProducer.send(textMessage); } // 如果設(shè)置了事務(wù)，會話就必須提交 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; // 實例化鏈接工廠，創(chuàng)建一個鏈接 connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKENURL); try { // 通過工廠獲取鏈接 connection = connectionFactory.createConnection(); // 啟動鏈接 connection.start(); // 創(chuàng)建會話，進行消息的接收 session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); // 創(chuàng)建消息隊列 destination = session.createQueue("talkWithMo"); // 創(chuàng)建一個消息的消費者 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. 測試

• 先運行生產(chǎn)者Producer

  
  圖7.png

ActiveMQ控制臺

圖8.png

• 再運行消費者Consumer

  
  圖9.png
  

  ActiveMQ控制臺

  
  圖10.png

V. 消息類型

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

2). request/reply模型

I. 實現(xiàn)思路

圖11.png

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

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

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

II. Server代碼

public class Server implements MessageListener { // 經(jīng)紀人鏈接 private static final String BROKER_URL = "tcp://localhost:61616"; // 請求隊列 private static final String REQUEST_QUEUE = "requestQueue"; // 經(jīng)紀人服務(wù) private BrokerService brokerService; // 會話 private Session session; // 生產(chǎn)者 private MessageProducer producer; // 消費者 private MessageConsumer consumer; private void start() throws Exception { createBroker(); setUpConsumer(); } /** * 創(chuàng)建經(jīng)紀人 * @throws Exception */ private void createBroker() throws Exception { // 創(chuàng)建經(jīng)紀人服務(wù) brokerService = new BrokerService(); // 設(shè)置是否持久化 brokerService.setPersistent(false); // 設(shè)置是否使用JMX brokerService.setUseJmx(false); // 添加鏈接 brokerService.addConnector(BROKER_URL); // 啟動 brokerService.start(); } /** * 設(shè)置消費者 * @throws JMSException */ private void setUpConsumer() throws JMSException { // 創(chuàng)建連接工廠 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL); // 創(chuàng)建連接 Connection connection = connectionFactory.createConnection(); // 啟動連接 connection.start(); // 創(chuàng)建Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 創(chuàng)建隊列 Destination adminQueue = session.createQueue(REQUEST_QUEUE); // 創(chuàng)建生產(chǎn)者 producer = session.createProducer(null); // 設(shè)置持久化模式 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 創(chuàng)建消費者 consumer = session.createConsumer(adminQueue); // 消費者設(shè)置消息監(jiān)聽 consumer.setMessageListener(this); } public void stop() throws Exception { producer.close(); consumer.close(); session.close(); brokerService.stop(); } @Override public void onMessage(Message message) { try { // 創(chuàng)建新消息 TextMessage response = this.session.createTextMessage(); // 判斷消息是否是文本消息 if (message instanceof TextMessage) { // 強轉(zhuǎn)為文本消息 TextMessage textMessage = (TextMessage) message; // 獲取消息內(nèi)容 String text = textMessage.getText(); // 設(shè)置消息 response.setText(handleRequest(text)); } response.setJMSCorrelationID(message.getJMSCorrelationID()); producer.send(message.getJMSReplyTo(), response); } catch (JMSException e) { e.printStackTrace(); } } /** * 構(gòu)建消息內(nèi)容 * @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 { // 經(jīng)紀人鏈接 private static final String BROKER_URL = "tcp://localhost:61616"; // 請求隊列 private static final String REQUEST_QUEUE = "requestQueue"; // 連接 private Connection connection; // 會話 private Session session; // 生產(chǎn)者 private MessageProducer producer; // 消費者 private MessageConsumer consumer; // 請求隊列 private Queue tempDest; public void start() throws JMSException { // 連接工廠 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(BROKER_URL); // 創(chuàng)建連接 connection = activeMQConnectionFactory.createConnection(); // 開啟連接 connection.start(); // 創(chuàng)建會話 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 創(chuàng)建隊列 Destination adminQueue = session.createQueue(REQUEST_QUEUE); // 創(chuàng)建生產(chǎn)者 producer = session.createProducer(adminQueue); // 設(shè)置持久化模式 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 創(chuàng)建模板隊列 tempDest = session.createTemporaryQueue(); // 創(chuàng)建消費者 consumer = session.createConsumer(tempDest); // 設(shè)置消息監(jiān)聽 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); // 創(chuàng)建文本消息 TextMessage textMessage = session.createTextMessage(); // 設(shè)置文本內(nèi)容 textMessage.setText(request); // 設(shè)置回復(fù) textMessage.setJMSReplyTo(tempDest); // 獲取UUID String correlationId = UUID.randomUUID().toString(); // 設(shè)置JMS id textMessage.setJMSCorrelationID(correlationId); // 發(fā)送消息 this.producer.send(textMessage); } @Override public 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

代碼下載

如果您覺得我的

作者：_凌浩雨
鏈接：https://www.jianshu.com/p/8b9bfe865e38
來源：簡書
簡書著作權(quán)歸作者所有，任何形式的轉(zhuǎn)載都請聯(lián)系作者獲得授權(quán)并注明出處。

轉(zhuǎn)載于:https://www.cnblogs.com/Jeely/p/11168287.html

總結(jié)

以上是生活随笔為你收集整理的分布式--ActiveMQ 消息中间件(一) https://www.jianshu.com/p/8b9bfe865e38的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。