jms什么意思(JMS是什么意思)

JMS即Java消息服務(Java Message Service)應用程序接口,是一個Java平臺中關于面向消息中間件(MOM)的API,用于在兩個應用程序之間,或分布式系統中發送消息,進行異步通信。Java消息服務是一個與具體平臺無關的API,絕大多數MOM提供商都對JMS提供支持。

當前環境

1. Mac OS 10.11.x
2. docker 1.12.1
3. JDK 1.8
4. SpringBoot 1.5

前言

基于之前一篇“一個故事告訴你什么是消息隊列”,了解了消息隊列的使用場景以及相關的特性。本文主要講述消息服務在 JAVA 中的使用。

市面上的有關消息隊列的技術選型非常多,如果我們的代碼框架要支持不同的消息實現,在保證框架具有較高擴展性的前提下,我們勢必要進行一定的封裝。

在 JAVA 中,大可不必如此。因為 JAVA 已經制定了一套標準的 JMS 規范。該規范定義了一套通用的接口和相關語義,提供了諸如持久、驗證和事務的消息服務,其最主要的目的是允許Java應用程序訪問現有的消息中間件。就和 JDBC 一樣。

基本概念

在介紹具體的使用之前,先簡單介紹一下 JMS 的一些基本知識。這里我打算分為 3 部分來介紹,即 消息隊列(MQ)的連接、消息發送與消息接收。

這里我們的技術選型是 SpringBoot、JMS、ActiveMQ

為了更好的理解 JMS,這里沒有使用 SpringBoot 零配置來搭建項目

MQ 的連接

使用 MQ 的第一步一定是先連接 MQ。因為這里使用的是 JMS 規范,對于任何遵守 JMS 規范的 MQ 來說,都會實現相應的ConnectionFactory接口,因此我們只需要創建一個ConnectionFactory工廠類,由它來實現 MQ 的連接,以及封裝一系列特性的 MQ 參數。

例子:這里我們以 ActiveMQ 為例,

maven 依賴:

org.springframework.bootspring-boot-starter-parent1.5.3.RELEAseorg.springframework.bootspring-boot-starter-activemq

創建 ActiveMQ 連接工廠:

@BeanpublicConnectionFactoryconnectionFactory(){ActiveMQConnectionFactoryconnectionFactory=newActiveMQConnectionFactory();connectionFactory.setBrokerURL(ActiveMQ_URL);connectionFactory.setUserName(ActiveMQ_USER);connectionFactory.setPassword(ActiveMQ_PASSWORD);returnconnectionFactory;}

消息發送

關于消息的發送,是通過 JMS 核心包中的JmsTemplate類來實現的,它簡化了 JMS 的使用,因為在發送或同步接收消息時它幫我們處理了資源的創建和釋放。從它的作用也不難推測出,它需要引用我們上面創建的連接工廠,具體代碼如下:

@BeanpublicJmsTemplatejmsQueueTemplate(){returnnewJmsTemplate(connectionFactory());}

JmsTemplate創建完成后,我們就可以調用它的方法來發送消息了。這里有兩個概念需要注意:

1. 消息會發送到哪里?-> 即需要指定發送隊列的目的地(Destination),是可以在 JNDI 中進行存儲和提取的 JMS 管理對象。
2. 發送的消息體具體是什么?-> 實現了javax.jms.Message的對象,類似于 JAVA RMI 的 Remote 對象。

代碼示例:

@AutowiredprivateJmsTemplatejmsQueueTemplate;/***發送原始消息Message*/publicvoidsend(){jmsQueueTemplate.send("queue1",newMessageCreator(){@OverridepublicMessagecreateMessage(Sessionsession)throwsJMSException{returnsession.createTextMessage("我是原始消息");}});}

優化:當然,我們不用每次都通過MessageCreator匿名類的方式來創建Message對象,JmsTemplate類中提供了對象實體自動轉換為Message對象的方法,convertAndSend(String destinationName, final Object message)。

優化代碼示例:

/***發送消息自動轉換成原始消息*/publicvoidconvertAndSend(){jmsQueueTemplate.convertAndSend("queue1","我是自動轉換的消息");}

注:關于消息轉換,還可以通過實現MessageConverter接口來自定義轉換內容

消息接收

講完了消息發送,我們最后來說說消息是如何接收的。消息既然是以Message對象的形式發送到指定的目的地,那么消息的接收勢必會去指定的目的地上去接收消息。這里采用的是監聽者的方式來監聽指定地點的消息,采用注解@JmsListener來設置監聽方法。

代碼示例:

@ComponentpublicclassListener1{@JmsListener(destination="queue1")publicvoidreceive(Stringmsg){System.out.println("監聽到的消息內容為:"+msg);}}

有了監聽的目標和方法后,監聽器還得和 MQ 關聯起來,這樣才能運作起來。這里的監聽器可能不止一個,如果每個都要和 MQ 建立連接,肯定不太合適。所以需要一個監聽容器工廠的概念,即接口JmsListenerContainerFactory,它會引用上面創建好的與 MQ 的連接工廠,由它來負責接收消息以及將消息分發給指定的監聽器。當然也包括事務管理、資源獲取與釋放和異常轉換等。

代碼示例:

@BeanpublicDefaultJmsListenerContainerFactoryjmsQueueListenerContainerFactory(){DefaultJmsListenerContainerFactoryfactory=newDefaultJmsListenerContainerFactory();factory.setConnectionFactory(connectionFactory());//設置連接數factory.setConcurrency("3-10");//重連間隔時間factory.setRecoveryInterval(1000L);returnfactory;}

場景

代碼地址:https://github.com/jasonGeng88/springboot-jms

對 JMS 有了基本的理解后,我們就來在具體的場景中使用一下。

首先,我們需要先啟動 ActiveMQ,這里我們以 Docker 容器化的方式進行啟動。

啟動命令:

dockerrun-d-p8161:8161-p61616:61616--nameactivemqwebcenter/activemq

啟動成功后,在 ActiveMQ 可視化界面查看效果(http://localhost:8161):

點對點模式(單消費者)

下面介紹消息隊列中最常用的一種場景,即點對點模式?；靖拍钊缦?

1. 每個消息只能被一個消費者(Consumer)進行消費。一旦消息被消費后,就不再在消息隊列中存在。
2. 發送者和接收者之間在時間上沒有依賴性,也就是說當發送者發送了消息之后,不管接收者有沒有正在運行,它不會影響到消息被發送到隊列。
3. 接收者在成功接收消息之后需向隊列應答成功。

代碼實現(為簡化代碼,部分代碼沿用上面所述): 啟動文件(Application.java)

@SpringBootApplication@EnableJmspublicclassApplication{.../***JMS隊列的模板類*connectionFactory()為ActiveMQ連接工廠*/@BeanpublicJmsTemplatejmsQueueTemplate(){returnnewJmsTemplate(connectionFactory());}publicstaticvoidmain(String[]args){SpringApplication.run(Application.class,args);}}

注解@EnableJms設置在@Configuration類上,用來聲明對 JMS 注解的支持。

消息生產者(PtpProducer.java)

@ComponentpublicclassPtpProducer{@AutowiredprivateJmsTemplatejmsQueueTemplate;/***發送消息自動轉換成原始消息*/publicvoidconvertAndSend(){jmsQueueTemplate.convertAndSend("ptp","我是自動轉換的消息");}}

生產者調用類(PtpController.java)

@RestController@RequestMapping(value="/ptp")publicclassPtpController{@AutowiredprivatePtpProducerptpProducer;@RequestMapping(value="/convertAndSend")publicObjectconvertAndSend(){ptpProducer.convertAndSend();return"success";}}

消息監聽容器工廠

@SpringBootApplication@EnableJmspublicclassApplication{.../***JMS隊列的監聽容器工廠*/@Bean(name="jmsQueueListenerCF")publicDefaultJmsListenerContainerFactoryjmsQueueListenerContainerFactory(){DefaultJmsListenerContainerFactoryfactory=newDefaultJmsListenerContainerFactory();factory.setConnectionFactory(connectionFactory());//設置連接數factory.setConcurrency("3-10");//重連間隔時間factory.setRecoveryInterval(1000L);returnfactory;}...}

消息監聽器

@ComponentpublicclassPtpListener1{/***消息隊列監聽器*destination隊列地址*containerFactory監聽器容器工廠,若存在2個以上的監聽容器工廠,需進行指定*/@JmsListener(destination="ptp",containerFactory="jmsQueueListenerCF")publicvoidreceive(Stringmsg){System.out.println("點對點模式1:"+msg);}}

演示

啟動項目啟動后,通過 REST 接口的方式來調用消息生產者發送消息,請求如下:

curl-XGET127.0.0.1:8080/ptp/convertAndSend

消費者控制臺信息:

ActiveMQ 控制臺信息:

列表說明:

• Name:隊列名稱。
• Number Of Pending Messages:等待消費的消息個數。
• Number Of Consumers:當前連接的消費者數目,因為我們采用的是連接池的方式連接,初始連接數為 3,所以顯示數字為 3。
• Messages Enqueued:進入隊列的消息總個數,包括出隊列的和待消費的,這個數量只增不減。
• Messages Dequeued:出了隊列的消息,可以理解為是已經消費的消息數量。

點對點模式(多消費者)

基于上面一個消費者消費的模式,因為生產者可能會有很多,同時像某個隊列發送消息,這時一個消費者可能會成為瓶頸。所以需要多個消費者來分攤消費壓力(消費線程池能解決一定壓力,但畢竟在單機上,做不到分布式分布,所以多消費者是有必要的),也就產生了下面的場景。

代碼實現

添加新的監聽器

@ComponentpublicclassPtpListener2{@JmsListener(destination=Constant.QUEUE_NAME,containerFactory="jmsQueueListenerCF")publicvoidreceive(Stringmsg){System.out.println("點對點模式2:"+msg);}}

演示 這里我們發起 10 次請求,來觀察消費者的消費情況:

這里因為監聽容器設置了線程池的緣故,在實際消費過程中,監聽器消費的順序會有所差異。

發布訂閱模式

除了點對點模式,發布訂閱模式也是消息隊列中常見的一種使用。試想一下,有一個即時聊天群,你在群里發送一條消息。所有在這個群里的人(即訂閱了該群的人),都會收到你發送的信息。

基本概念:

1. 每個消息可以有多個消費者。
2. 發布者和訂閱者之間有時間上的依賴性。針對某個主題(Topic)的訂閱者,它必須創建一個訂閱者之后,才能消費發布者的消息。
3. 為了消費消息,訂閱者必須保持運行的狀態。

代碼實現 修改 JmsTemplate 模板類,使其支持發布訂閱功能

@SpringBootApplication@EnableJmspublicclassApplication{...@BeanpublicJmsTemplatejmsTopicTemplate(){JmsTemplatejmsTemplate=newJmsTemplate(connectionFactory());jmsTemplate.setPubSubDomain(true);returnjmsTemplate;}...}

消息生產者(PubSubProducer.java)

@ComponentpublicclassPtpProducer{@AutowiredprivateJmsTemplatejmsTopicTemplate;publicvoidconvertAndSend(){jmsTopicTemplate.convertAndSend("topic","我是自動轉換的消息");}}

生產者調用類(PubSubController.java)

@RestController@RequestMapping(value="/pubsub")publicclassPtpController{@AutowiredprivatePubSubProducerpubSubProducer;@RequestMapping(value="/convertAndSend")publicStringconvertAndSend(){pubSubProducer.convertAndSend();return"success";}}

修改 DefaultJmsListenerContainerFactory 類,使其支持發布訂閱功能

@SpringBootApplication@EnableJmspublicclassApplication{.../***JMS隊列的監聽容器工廠*/@Bean(name="jmsTopicListenerCF")publicDefaultJmsListenerContainerFactoryjmsTopicListenerContainerFactory(){DefaultJmsListenerContainerFactoryfactory=newDefaultJmsListenerContainerFactory();factory.setConnectionFactory(connectionFactory());factory.setConcurrency("1");factory.setPubSubDomain(true);returnfactory;}...}

消息監聽器(這里設置2個訂閱者)

@ComponentpublicclassPubSubListener1{@JmsListener(destination="topic",containerFactory="jmsTopicListenerCF")publicvoidreceive(Stringmsg){System.out.println("訂閱者1-"+msg);}}@ComponentpublicclassPubSubListener2{@JmsListener(destination="topic",containerFactory="jmsTopicListenerCF")publicvoidreceive(Stringmsg){System.out.println("訂閱者2-"+msg);}}

演示

curl-XGET127.0.0.1:8080/pubSub/convertAndSend

消費者控制臺信息:

ActiveMQ 控制臺信息: