基于JMS(Java Message Service)的消息中間件----ActiveMQ

JessYanCoding 發布于2019-07-24 10:41 / 2095人閱讀

摘要：而中的消息中間件則是在常見的消息中間件類型無疑是不錯的選擇。是在之間傳遞的消息的對象。基本功能是用于和面向消息的中間件相互通信的應用程序接口。支持兩種消息發送和接收模型。一種稱為模型，即采用點對點的方式發送消息。

消息中間件利用高效可靠的消息傳遞機制進行平臺無關的數據交流，并基于數據通信來進行分布式系統的集成。通過提供消息傳遞和消息排隊模型，它可以在分布式環境下擴展進程間的通信。而Java中的消息中間件則是JMS---Java Message Service.在常見的消息中間件類型ActiveMQ無疑是不錯的選擇。接下來我們先簡單介紹一下什么是JMS及ActiveMQ而后我們介紹一下ActiveMQ中的兩種模式 -- 隊列模型以及主題模式。

JMS簡介

JMS基本概念
JMS(Java Message Service)是訪問企業消息系統的標準API,它便于消息系統中的Java應用程序進行消息交換,并且通過提供標準的產生、發送、接收消息的接口簡化企業應用的開發。

JMS應用由以下幾部分組成：
JMS provider ：是一個消息系統，它實現了JMS 接口并提供管理和控制的功能。
JMS clients ：是用Java語言寫的一些程序和組件，它們產生和使用消息。
Messages ：是在JMS clients之間傳遞的消息的對象。
Administered objects ：是由使用JMS clients 的人生成的預選設置好的JMS 對象。有兩種這樣的對象：
destinations和connection factories。

JMS基本功能
JMS是用于和面向消息的中間件相互通信的應用程序接口。它既支持點對點(point-to-point)的域，又支持發布/訂閱 (publish/subscribe)類型的域，并且提供對下列類型的支持：經認可的消息傳遞,事務型消息的傳遞，一致性消息和具有持久性的訂閱者支持。JMS還提供了另一種方式來對您的應用與舊的后臺系統相集成。

消息服務類型
1) point-to-point (PTP)方式：點到點的模型。消息由一個JMS客戶機（發布者）發送到服務器上的一個目的地，即一個隊列（queue）。而另一個JMS客戶機（訂閱者）則可以訪問這個隊列，并從該服務器獲取這條消息。
2) publish/subscribe (pub/sub)方式：發布-訂閱模型。這里仍然是由一個JMS客戶機將一條消息發布到服務器上的一個目的地上，但是這次這個目的地叫做一個主題（topic），可有多個訂閱者去訪問該消息。消息將一直維持在主題中，直到這個主題的所有訂閱者都取走了該消息的一個副本。消息也包括了一個參數，用于定義了該消息的耐久性（它能夠在服務器上等待訂閱者多長時間）。

ActiceMQ簡介：

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

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

Java實現ActiveMQ兩種模型

【1】隊列模型

隊列模型之生產者端

package com.imooc.jms.queue;
import java.awt.font.TextMeasurer;
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 AppProducer {

   private static final String url = "tcp://192.168.133.1:61616";
   private static final String queueName = "queue-test";
   
   public static void main(String[] args) throws JMSException {
       
       //1.創建connectionFactory
       ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
       
       //2.創建connection
       Connection connection = connectionFactory.createConnection();
       
       //3.  啟動鏈接
       connection.start();
       
       //4. 創建會話
       Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
       
       //5.  創建目標
       Destination destination =  session.createQueue(queueName);
       
       //6.  創建生產者向目標發送消息
       MessageProducer  producer=  session.createProducer(destination);
       
       for(int i = 0 ;i<100;i++){
           //7.  創建消息
           TextMessage textMessage = session.createTextMessage("test"+i);
           //8.  發布消息
           producer.send(textMessage);
           System.out.println("發送消息："+textMessage.getText());
       }    
       //9. 關閉連接
       connection.close();
   }

}

說明：此為隊列模型中的生產者端，下面就該端代碼進行說明：
new 一個ConnectionFactory --》 new 一個connection ---》啟動這個連接----》創建一個會話----》創建消息發送目的地---》創建一個消息生產者向目的地發送消息--》發送消息----》關閉連接

隊列模型之消費者端

package com.imooc.jms.queue;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage；
import org.apache.activemq.ActiveMQConnectionFactory;

public class AppConsumer {

   private static final String url = "tcp://192.168.133.1:61616";
   private static final String queueName = "queue-test";
   
   public static void main(String[] args) throws JMSException {
       
       //1.創建connectionFactory
       ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
       
       //2.創建connection
       Connection connection = connectionFactory.createConnection();
       
       //3.  啟動鏈接
       connection.start();
       
       //4. 創建會話
       Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
       
       //5.  創建目標
       Destination destination =  session.createQueue(queueName);
       
      //6.  創建一個消費者
       MessageConsumer consumer =  session.createConsumer(destination);
       
       //7.  創建一個監聽器
       consumer.setMessageListener(new MessageListener() {
           
           public void onMessage(Message message) {
               
               TextMessage textMessage = (TextMessage)message;
               try {
                   System.out.println("接受消息："+textMessage.getText());
               } catch (JMSException e) {
                   // TODO Auto-generated catch block
                   e.printStackTrace();
               }
               
           }
       });
       
       //9. 關閉連接
       //connection.close();
   }

}

說明：此為隊列模型中的消費者端，下面就該端代碼進行說明：
new 一個ConnectionFactory --》 new 一個connection ---》啟動這個連接----》創建一個會話----》創建消息發送目的地---》創建一個消息消費者--》創建一個消息監聽器監聽消息----》關閉連接

【2】主題模型

主題模型之生產者端

package com.imooc.jms.topic;
import java.awt.font.TextMeasurer;
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 AppProducer {

   private static final String url = "tcp://192.168.133.1:61616";
   private static final String topicName = "topic-test";
   
   public static void main(String[] args) throws JMSException {
       
       //1.創建connectionFactory
       ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
       
       //2.創建connection
       Connection connection = connectionFactory.createConnection();
       
       //3.  啟動鏈接
       connection.start();
       
       //4. 創建會話
       Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
       
       //5.  創建目標
       Destination destination =  session.createTopic(topicName);
       
       //6.  創建生產者向目標發送消息
       MessageProducer  producer=  session.createProducer(destination);
       
       for(int i = 0 ;i<100;i++){
           //7.  創建消息
           TextMessage textMessage = session.createTextMessage("test"+i);
           //8.  發布消息
           producer.send(textMessage);
           System.out.println("發送消息："+textMessage.getText());
       }
       
       //9. 關閉連接
       connection.close();
   }

}

主題模型之消費者端

package com.imooc.jms.topic;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class AppConsumer {

   private static final String url = "tcp://192.168.133.1:61616";
   private static final String topicName = "topic-test";
   
   public static void main(String[] args) throws JMSException {
       
       //1.創建connectionFactory
       ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
       
       //2.創建connection
       Connection connection = connectionFactory.createConnection();
       
       //3.  啟動鏈接
       connection.start();
       
       //4. 創建會話
       Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
       
       //5.  創建目標
       Destination destination =  session.createTopic(topicName);
       
      //6.  創建一個消費者
       MessageConsumer consumer =  session.createConsumer(destination);
       
       //7.  創建一個監聽器
       consumer.setMessageListener(new MessageListener() {
           
           public void onMessage(Message message) {
               
               TextMessage textMessage = (TextMessage)message;
               try {
                   System.out.println("接受消息："+textMessage.getText());
               } catch (JMSException e) {
                   // TODO Auto-generated catch block
                   e.printStackTrace();
               }
               
           }
       });
       
       //9. 關閉連接
       //connection.close();
   }

}

最后創建項目源碼地址：
鏈接：http://pan.baidu.com/s/1bo1PSib 密碼：khyc

GPU云服務器云服務器消息中間件activemq activemq怎么發送消息給中間件 java消息中間件 java消息隊列中間件

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://m.specialneedsforspecialkids.com/yun/35878.html

慕課網_《Java消息中間件》學習總結

摘要：時間年月日星期六說明本文部分內容均來自慕課網。這個時候，可以啟動多臺積分系統，來同時消費這個消息中間件里面的登錄消息，達到橫向擴展的作用。時間：2017年07月22日星期六說明：本文部分內容均來自慕課網。@慕課網：http://www.imooc.com教學源碼：無學習源碼：https://github.com/zccodere/s... 第一章：課程介紹 1-1 課程安排 Java...

twohappy 2019-08-15 11:00 評論0 收藏0
消息中間件及ActiveMQ介紹

摘要：中間件的分類基于遠程過程調用的中間件。基于對象請求代理的中間件。消息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用于諸如遠程過程調用的技術。一.中間件 1.1 什么是中間件? 由于業務、機構和技術是不斷變化的，因此為其服務的軟件系統必須適應這樣的變化。在合并、添加服務或擴展可用服務之后，公司可能無力負擔重新創建信息系統所需的成本。正是在...

jaysun 2019-08-15 17:48 評論0 收藏0
新手也能看懂，消息隊列其實很簡單

摘要：通過以上分析我們可以得出消息隊列具有很好的削峰作用的功能即通過異步處理，將短時間高并發產生的事務消息存儲在消息隊列中，從而削平高峰期的并發事務。該文已加入開源項目：JavaGuide（一份涵蓋大部分Java程序員所需要掌握的核心知識的文檔類項目，Star 數接近 16k）。地址:https://github.com/Snailclimb... 本文內容思維導圖：showImg(ht...

Clect 2019-08-16 14:01 評論0 收藏0
消息隊列ActiveMQ的使用詳解

摘要：學習消息隊列的使用之前，我們先來搞清。是操作消息的接口。消息生產者由創建，并用于將消息發送到。接收消息打印結果這是接收到的消息消費者啟動。。。。通過上一篇文章《消息隊列深入解析》，我們已經消息隊列是什么、使用消息隊列的好處以及常見消息隊列的簡單介紹。這一篇文章，主要帶大家詳細了解一下消息隊列ActiveMQ的使用。學習消息隊列ActiveMQ的使用之前，我們先來搞清JMS。 J...

niceforbear 2019-08-15 15:40 評論0 收藏0