ActiveMQ知識概括

• ActiveMQ簡介
• Java實現(xiàn)ActiveMQ
• JMS規(guī)范與落地
• ActiveMQ的broker
• Spring，SpringBoot整合ActiveMQ
• ActiveMQ的傳輸協(xié)議
• ActiveMQ的消息存儲和持久化
• ActiveMQ多節(jié)點集群
• ActiveMQ高級特性

ActiveMQ簡介

ActiveMQ安裝：

• 安裝步驟：
  ①去ActiveMQ官網(wǎng)下載壓縮包。
  ②解壓壓縮包到指定目錄。
  ③啟動ActiveMQ：service activemq start
  ④查看activemq狀態(tài)：service activemq status
  ⑤關閉activemq服務：service activemq stop
• 啟動時指定日志輸出文件：
  ①activemq日志默認的位置是在：%activemq安裝目錄%/data/activemq.log
  ②這是我們啟動時指定日志輸出文件：service activemq start > /usr/local/raohao/activemq.log
• 查看程序啟動是否成功的3種方式（通用）：
  ①ps -ef | grep activemq
  ②netstat -anp | grep 61616
  ③lsof -i： 61616

ActiveMQ控制臺：

• 訪問activemq管理頁面地址：http://IP地址:8161/。默認的用戶名和密碼是admin/admin。
• 備注：
  ①ActiveMQ采用61616端口提供JMS服務。
  ②ActiveMQ采用8161端口提供管理控制臺服務。
• 默認程序連接activemq(JMS服務)是不需要密碼的，為了安裝起見，一般都會設置密碼，提高安全性。
• ActiveMQ控制臺之隊列：
  ①Number Of Pending Messages：等待消費的消息，這個是未出隊列的數(shù)量，公式=總接收數(shù)-總出隊列數(shù)。
  ②Number Of Consumers：消費者數(shù)量，消費者端的消費者數(shù)量。
  ③Messages Enqueued：進隊消息數(shù)，進隊列的總消息量，包括出隊列的。這個數(shù)只增不減。
  ④Messages Dequeued：出隊消息數(shù)，可以理解為是消費者消費掉的數(shù)量。
  
• ActiveMQ控制臺之主題：
  
• ActiveMQ控制臺之訂閱者：
  

Java實現(xiàn)ActiveMQ

pom.xml導入依賴：

<!-- activemq 所需要的jar 包--> <dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.15.9</version> </dependency> <!-- activemq 和 spring 整合的基礎包 --> <dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId><version>3.16</version> </dependency>

JMS編碼總體規(guī)范：

• 架構(gòu)：
  
• JMS開發(fā)的基本步驟：
  ①創(chuàng)建一個connection factory
  ②通過connection factory來創(chuàng)建JMS connection
  ③啟動JMS connection
  ④通過connection創(chuàng)建JMS session
  ⑤創(chuàng)建JMS destination
  ⑥創(chuàng)建JMS producer或者創(chuàng)建JMS message并設置destination
  ⑦創(chuàng)建JMS consumer或者是注冊一個JMS message listener
  ⑧發(fā)送或者接受JMS message(s)
  ⑨關閉所有的JMS資源(connection, session, producer, consumer等)

Destination簡介：

• Destination是目的地。下面拿jvm和mq，做個對比。目的地，我們可以理解為是數(shù)據(jù)存儲的地方。
  
• Destination分為兩種：隊列和主題。
  ①在點對點的消息傳遞域中,目的地被稱為隊列(queue)
  ②在發(fā)布訂閱消息傳遞域中,目的地被稱為主題(topic)
  ③下圖介紹：
  

隊列消息（Queue）總結(jié)：

• 兩種消費方式：
  ①同步阻塞方式(receive)：訂閱者或接收者抵用MessageConsumer的receive()方法來接收消息，receive方法在能接收到消息之前（或超時之前）將一直阻塞。
  ②異步非阻塞方式（監(jiān)聽器onMessage()）：訂閱者或接收者通過MessageConsumer的setMessageListener(MessageListener listener)注冊一個消息監(jiān)聽器，當消息到達之后，系統(tǒng)會自動調(diào)用監(jiān)聽器MessageListener的onMessage(Message message)方法。
• 隊列的特點：
  ①每個消息只能有一個消費者，類似1對1的關系。好比個人快遞自己領取自己的。
  ②消息的生產(chǎn)者和消費者之間沒有時間上的相關性。無論消費者在生產(chǎn)者發(fā)送消息的時候是否處于運行狀態(tài)，消費者都可以提取消息。好比我們的發(fā)送短信，發(fā)送者發(fā)送后不見得接收者會即收即看。
  ③消息被消費后隊列中不會再存儲，所以消費者不會消費到已經(jīng)被消費掉的消息。
• 消息消費情況：
  ①情況1：只啟動消費者1。結(jié)果：消費者1會消費所有的數(shù)據(jù)。
  ②情況2：先啟動消費者1，再啟動消費者2。結(jié)果：消費者1消費所有的數(shù)據(jù)。消費者2不會消費到消息。
  ③情況3：生產(chǎn)者發(fā)布6條消息，在此之前已經(jīng)啟動了消費者1和消費者2。結(jié)果：消費者1和消費者2平攤了消息。各自消費3條消息。
  ④疑問：怎么去將消費者1和消費者2不平均分攤呢？而是按照各自的消費能力去消費。我覺得，現(xiàn)在activemq就是這樣的機制。

主題消息(Topic)介紹：

• 在發(fā)布訂閱消息傳遞域中，目的地被稱為主題（topic）
• 發(fā)布/訂閱消息傳遞域的特點如下：
  ①生產(chǎn)者將消息發(fā)布到topic中，每個消息可以有多個消費者，屬于1：N的關系；
  ②生產(chǎn)者和消費者之間有時間上的相關性。訂閱某一個主題的消費者只能消費自它訂閱之后發(fā)布的消息。
  ③生產(chǎn)者生產(chǎn)時，topic不保存消息它是無狀態(tài)的不落地，假如無人訂閱就去生產(chǎn)，那就是一條廢消息，所以，一般先啟動消費者再啟動生產(chǎn)者。
  ④默認情況下如上所述，但是JMS規(guī)范允許客戶創(chuàng)建持久訂閱，這在一定程度上放松了時間上的相關性要求。持久訂閱允許消費者消費它在未處于激活狀態(tài)時發(fā)送的消息。一句話，好比我們的微信公眾號訂閱

tpoic和queue對比：

比較項目Topic模式隊列Queue模式隊列

工作模式.	“訂閱-發(fā)布"模式，如果當前沒有訂閱者，消息將會被丟棄。如果有多個訂閱者，那么這些訂閱者都會收到消息	“負載均衡"模式，如果當前沒有消費者，消息也不會云棄;如果有多個消費者，那么—條消息也只會發(fā)送始其中一個消費者，并且要求消費者ack信息
有無狀態(tài)	無狀態(tài)	Queue數(shù)據(jù)默認會在mq服務器上以文件形式保存，比如Active MQ—般保存在SAMQ_HOME\datakr-storeldata下面。也可以配置成DB存儲。
傳遞完整性	如果沒有訂閱者，消息會被丟棄	消息不會云棄
處理效率	由于消息要按照訂閱者的數(shù)量進行復制，所以處理性能會隨著訂閱者的增加而明顯降低,并且還要結(jié)合不同消息協(xié)議自身的性能差異	由于—條消息只發(fā)送給—個消費者，所以就算消費者再多，性能也不會有明顯降低。當然不同消息協(xié)議的具體性能也是有差異的

JMS規(guī)范與落地

JMS是什么：

• JMS是Java消息服務
• Java消息服務指的是兩個應用程序之間進行異步通信的API，它為標準協(xié)議和消息服務提供了一組通用接口，包括創(chuàng)建、發(fā)送、讀取消息等，用于支持Java應用程序開發(fā)。在JavaEE中，當兩個應用程序使用JMS進行通信時，它們之間不是直接相連的，而是通過一個共同的消息收發(fā)服務組件關聯(lián)起來以達到解耦/異步削峰的效果。

JMS的組成結(jié)構(gòu)和特點：

消息頭：

• JMS的消息頭有哪些屬性：
  ①JMSDestination：消息目的地
  ②JMSDeliveryMode：消息持久化模式
  ③JMSExpiration：消息過期時間
  ④JMSPriority：消息的優(yōu)先級
  ⑤JMSMessageID：消息的唯一標識符。后面我們會介紹如何解決冪等性。
• 說明： 消息的生產(chǎn)者可以set這些屬性，消息的消費者可以get這些屬性。這些屬性在send方法里面也可以設置。

消息體：

• 封裝具體的消息數(shù)據(jù)
• 5種消息體格式：
  ①TextMessage——普通字符串消息，包含一個string
  ②MapMessage——一個Map類型的消息，key為string類型，而值為Java的基本類型
  ③BytesMessage——二進制數(shù)組消息，包含一個byte[]
  ④StreamMessage——Java數(shù)據(jù)流消息，用標準流操作來順序的填充和讀取。
  ⑤ObjectMessage——對象消息，包含一個可序列化的Java對象
• 發(fā)送和接受的消息體類型必須一致對應

消息屬性：

• 如果需要除消息頭字段之外的值，那么可以使用消息屬性。他是識別/去重/重點標注等操作，非常有用的方法。
• 他們是以屬性名和屬性值對的形式制定的。可以將屬性是為消息頭得擴展，屬性指定一些消息頭沒有包括的附加信息，比如可以在屬性里指定消息選擇器。消息的屬性就像可以分配給一條消息的附加消息頭一樣。它們允許開發(fā)者添加有關消息的不透明附加信息。它們還用于暴露消息選擇器在消息過濾時使用的數(shù)據(jù)。
• 下圖是設置消息屬性的API：set對應類型Property(String name，對應類型 value)

JMS的可靠性：

• PERSISTENT：持久性
• Transaction：事務
• Acknowledge：簽收

消息的持久化：

• 什么是持久化消息？
  ①保證消息只被傳送一次和成功使用一次。在持久性消息傳送至目標時，消息服務將其放入持久性數(shù)據(jù)存儲。如果消息服務由于某種原因?qū)е率?#xff0c;它可以恢復此消息并將此消息傳送至相應的消費者。雖然這樣增加了消息傳送的開銷，但卻增加了可靠性。
  ②我的理解：在消息生產(chǎn)者將消息成功發(fā)送給MQ消息中間件之后。無論是出現(xiàn)任何問題，如：MQ服務器宕機、消費者掉線等。都保證（topic要之前注冊過，queue不用）消息消費者，能夠成功消費消息。如果消息生產(chǎn)者發(fā)送消息就失敗了，那么消費者也不會消費到該消息。
• 參數(shù)設置說明：
  ①非持久：非持久化：當服務器宕機，消息不存在。
  messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)
  ②持久：持久化：當服務器宕機，消息依然存在。
  messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT)
  ③Queue默認是持久。
• 持久的Queue：持久化消息這是隊列的默認傳遞模式，此模式保證這些消息只被傳送一次和成功使用一次。對于這些消息，可靠性是優(yōu)先考慮的因素。可靠性的另一個重要方面是確保持久性消息傳送至目標后，消息服務在向消費者傳送它們之前不會丟失這些消息。
• 持久的Topic：一定要先運行一次消費者,類似于像MQ注冊,我訂閱了這個主題。然后再運行主題生產(chǎn)者，無論消費著是否在線,都會接收到,在線的立即接收到,不在線的等下次上線把沒接收到的接收。類似微信公眾號訂閱發(fā)布。

消息事務：

• producer提交時的事務：
  ①false：只要執(zhí)行send，就進入到隊列中，關閉事務，那第2個簽收參數(shù)的設置需要有效。
  ②true：先執(zhí)行send再執(zhí)行commit，消息才被真正提交到隊列中，消息需要需要批量提交，需要緩沖處理。
• consumer消費時的事務：
  ①false：activeMQ默認認為你執(zhí)行了commit，消費了消息。
  ②true：只有執(zhí)行了commit，activeMQ才認為你消費了消息，控制臺的消費數(shù)才會上升。不執(zhí)行commit的話，會重復消費消息！
• 事務偏生產(chǎn)者/簽收偏消費者！

消息簽收：

• 非事務：
  ①自動簽收（Session.AUTO_ACKNOWLEDGE）：該方式是默認的。該種方式，無需我們程序做任何操作，框架會幫我們自動簽收收到的消息。
  ②手動簽收（Session.CLIENT_ACKNOWLEDGE）：手動簽收。該種方式，需要我們手動調(diào)用Message.acknowledge()，來簽收消息。如果不簽收消息，該消息會被我們反復消費，只到被簽收。
  ③允許重復消息（Session.DUPS_OK_ACKNOWLEDGE）：多線程或多個消費者同時消費到一個消息，因為線程不安全，可能會重復消費。該種方式很少使用到。
  ④事務下的簽收（Session.SESSION_TRANSACTED）：開始事務的情況下，可以使用該方式。該種方式很少使用到。
• 事務：
  ①由于消費者開啟了事務,沒有提交事務(就算手動簽收也沒用),服務器認為,消費者沒有收到消息。
  ②生產(chǎn)事務開啟，只有commit后才能將全部消息變?yōu)橐严M。
• 簽收和事務的關系：
  ①在事務性會話中，當一個事務被成功提交則消息被自動簽收。如果事務回滾，則消息會被再次傳送。事務優(yōu)先于簽收，開始事務后，簽收機制不再起任何作用。
  ②非事務性會話中，消息何時被確認取決于創(chuàng)建會話時的應答模式。
  ③消費者事務開啟，只有commit后才能將全部消息變?yōu)橐严M。
  ④事務偏向生產(chǎn)者，簽收偏向消費者。也就是說生產(chǎn)者使用事務更好點，消費者使用簽收機制更好點。

JMS的點對點總結(jié)：

• 點對點模型是基于隊列的，生產(chǎn)者發(fā)送消息到隊列，消費者從隊列接收消息，隊列的存在使得消息的異步傳輸成為可能。和我們平時給朋友發(fā)送短信類似。
  ①如果在Session關閉時有部分消息被收到但還沒有被簽收（acknowledge），那當消費者下次連接到相同的隊列時，這些消息還會被再次接收。
  ②隊列可以長久的保存消息直到消費者收到消息。消費者不需要因為擔心消息會丟失而時刻和隊列保持激活的鏈接狀態(tài)，充分體現(xiàn)了異步傳輸模式的優(yōu)勢

JMS的發(fā)布訂閱總結(jié)：

• 非持久訂閱：
  ①非持久訂閱只有當客戶端處于激活狀態(tài)，也就是和MQ保持連接狀態(tài)才能收發(fā)到某個主題的消息。如果消費者處于離線狀態(tài)，生產(chǎn)者發(fā)送的主題消息將會丟失作廢，消費者永遠不會收到。一句話：先訂閱注冊才能接受到發(fā)布，只給訂閱者發(fā)布消息。
• 持久訂閱：
  ①客戶端首先向MQ注冊一個自己的身份ID識別號，當這個客戶端處于離線時，生產(chǎn)者會為這個ID保存所有發(fā)送到主題的消息，當客戶再次連接到MQ的時候，會根據(jù)消費者的ID得到所有當自己處于離線時發(fā)送到主題的消息當持久訂閱狀態(tài)下，不能恢復或重新派送一個未簽收的消息。持久訂閱才能恢復或重新派送一個未簽收的消息。
• 用哪個?
  ①當所有的消息必須被接收，則用持久訂閱。當消息丟失能夠被容忍，則用非持久訂閱。

ActiveMQ的broker

簡介：

• 相當于一個ActiveMQ服務器實例說白了，Broker其實就是實現(xiàn)了用代碼的形式啟動ActiveMQ將MQ嵌入到Java代碼中，以便隨時用隨時啟動，在用的時候再去啟動這樣能節(jié)省了資源，也保證了可用性。

嵌入式Broker：

• POM.XML：

<dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.15.11</version> </dependency> <dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.10.1</version> </dependency> <dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId><version>4.15</version> </dependency>

• 主啟動類：

import org.apache.activemq.broker.BrokerService; public class EmbedBroker { public static void main(String[] args) throws Exception { //ActiveMQ也支持在vm中通信基于嵌入的broker BrokerService brokerService = new BrokerService(); brokerService.setPopulateJMSXUserID(true); brokerService.addConnector("tcp://127.0.0.1:61616"); brokerService.start(); } }

• 和Linux上的ActiveMQ是一樣的,Broker相當于一個Mini版本的ActiveMQ

Spring，SpringBoot整合ActiveMQ

Spring整合ActiveMQ：

• Maven修改，需要添加Spring支持JMS的包：

<!-- activemq核心依賴包 --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.10.0</version></dependency><!-- 嵌入式activemq的broker所需要的依賴包 --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.10.1</version></dependency><!-- activemq連接池 --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId><version>5.15.10</version></dependency><!-- spring支持jms的包 --><dependency><groupId>org.springframework</groupId><artifactId>spring-jms</artifactId><version>5.2.1.RELEASE</version></dependency><!--spring相關依賴包--><dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId><version>4.15</version></dependency>

• Spring配置文件：

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd"><!-- 開啟包的自動掃描 --><context:component-scan base-package="com.activemq.demo"/><!-- 配置生產(chǎn)者 --><bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"><property name="connectionFactory"><!-- 正真可以生產(chǎn)Connection的ConnectionFactory,由對應的JMS服務商提供 --><bean class="org.apache.activemq.spring.ActiveMQConnectionFactory"><property name="brokerURL" value="tcp://192.168.10.130:61616"/></bean></property><property name="maxConnections" value="100"/></bean><!-- 這個是隊列目的地,點對點的Queue --><bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue"><!-- 通過構(gòu)造注入Queue名 --><constructor-arg index="0" value="spring-active-queue"/></bean><!-- 這個是隊列目的地, 發(fā)布訂閱的主題Topic--><bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic"><constructor-arg index="0" value="spring-active-topic"/></bean><!-- Spring提供的JMS工具類,他可以進行消息發(fā)送,接收等 --><bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"><!-- 傳入連接工廠 --><property name="connectionFactory" ref="connectionFactory"/><!-- 傳入目的地 --><property name="defaultDestination" ref="destinationQueue"/><!-- 消息自動轉(zhuǎn)換器 --><property name="messageConverter"><bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/></property></bean> </beans>

• 隊列（Queue）：

---------------生產(chǎn)者------------------ @Service public class SpringMQ_Producer { private JmsTemplate jmsTemplate; @Autowired public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } public static void main(String[] args) { ApplicationContext applicationContext = new ClassPathXmlApplicationContext("Application.xml"); SpringMQ_Producer springMQ_producer = applicationContext.getBean(SpringMQ_Producer.class); springMQ_producer.jmsTemplate.send(session -> session.createTextMessage("***Spring和ActiveMQ的整合case111.....")); System.out.println("********send task over"); } } ---------------消費者------------------ @Service public class SpringMQ_Consumer { private JmsTemplate jmsTemplate; @Autowired public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } public static void main(String[] args) { ApplicationContext applicationContext = new ClassPathXmlApplicationContext("Application.xml"); SpringMQ_Consumer springMQ_consumer = applicationContext.getBean(SpringMQ_Consumer.class); String returnValue = (String) springMQ_consumer.jmsTemplate.receiveAndConvert(); System.out.println("****消費者收到的消息: " + returnValue); } }

• 主題（Topic）：

---------------生產(chǎn)者------------------ @Service public class SpringMQ_Topic_Producer { private JmsTemplate jmsTemplate; public SpringMQ_Topic_Producer(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } public static void main(String[] args) { ApplicationContext applicationContext = new ClassPathXmlApplicationContext("Application.xml"); SpringMQ_Topic_Producer springMQ_topic_producer = applicationContext.getBean(SpringMQ_Topic_Producer.class); //直接調(diào)用application.xml里面創(chuàng)建的destinationTopic這個bean設置為目的地就行了 springMQ_topic_producer.jmsTemplate.setDefaultDestination(((Destination) applicationContext.getBean("destinationTopic"))); springMQ_topic_producer.jmsTemplate.send(session -> session.createTextMessage("***Spring和ActiveMQ的整合TopicCase111.....")); } } ---------------消費者------------------ @Service public class SpringMQ_Topic_Consumer { private JmsTemplate jmsTemplate; public SpringMQ_Topic_Consumer(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } public static void main(String[] args) { ApplicationContext applicationContext = new ClassPathXmlApplicationContext("Application.xml"); SpringMQ_Topic_Consumer springMQConsumer = applicationContext.getBean(SpringMQ_Topic_Consumer.class); //直接調(diào)用application.xml里面創(chuàng)建的destinationTopic這個bean設置為目的地就行了 springMQConsumer.jmsTemplate.setDefaultDestination(((Destination) applicationContext.getBean("destinationTopic"))); String returnValue = (String) springMQConsumer.jmsTemplate.receiveAndConvert(); System.out.println("****消費者收到的消息: " + returnValue); } }

• 在Spring里面實現(xiàn)消費者不啟動，直接通過配置監(jiān)聽完成：

<!--/配置監(jiān)聽程序--> <bean id="jmscontainer" class="org.springframework.jms.1listener.DefaultlessageListenerContainer"><property name="connectionFactory" ref="jmsFactory" /><property name="destination" ref="destinationTopic" /><!-- public class MyMessageListener implements MessageListener--><property name="messageListener" ref="myMessageListener" /> </bean> //實現(xiàn)MessageListener的類,需要把這個類交給xml配置里面的DefaultMessageListenerContainer管理 @Component public class MyMessageListener implements MessageListener { @Override public void onMessage(Message message) { if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; try { System.out.println("消費者收到的消息" + textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } } }

SpringBoot整合ActiveMQ：

• POM文件：

<!--spring boot整合activemq的jar包--> <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-activemq</artifactId><version>2.1.5.RELEASE</version> </dependency>

• YML文件：

# web占用的端口 server:port: 7777spring:activemq:# activemq的broker的urlbroker-url: tcp://192.168.17.3:61616# 連接activemq的broker所需的賬號和密碼user: adminpassword: adminjms:# 目的地是queue還是topic， false（默認） = queue true = topicpub-sub-domain: false# 自定義隊列名稱。這只是個常量 myQueueName: springboot-activemq-queue # 自定義主題名稱。這只是個常量 myTopicName: springboot-activemq-topic

• 配置bean：

@Component @EnableJms //開啟Springboot的Jms public class ConfigBean { @Value("myQueueName") private String myQueueName; @Bean public ActiveMQQueue queue() { //創(chuàng)建一個ActiveMQQueue return new ActiveMQQueue(myQueueName); }@Value("${myTopicName}") private String topicName; @Bean public ActiveMQTopic activeMQTopic() { //創(chuàng)建一個ActiveMQTopicreturn new ActiveMQTopic(topicName); } }

• 隊列(queue)：

-------------生產(chǎn)者------------- @Component public class Queue_Produce {// JMS模板@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate ;// 這個是我們配置的隊列目的地@Autowiredprivate Queue queue ;// 發(fā)送消息public void produceMessage(){// 一參是目的地，二參是消息的內(nèi)容jmsMessagingTemplate.convertAndSend(queue,"****"+ UUID.randomUUID().toString().substring(0,6));}// 定時任務。每3秒執(zhí)行一次。非必須代碼，僅為演示。@Scheduled(fixedDelay = 3000)public void produceMessageScheduled(){produceMessage();} } -------------消費者------------- @Component public class Queue_consummer {// 注冊一個監(jiān)聽器。destination指定監(jiān)聽的主題。@JmsListener(destination = "${myqueue}")public void receive(TextMessage textMessage) throws Exception{System.out.println(" *** 消費者收到消息 ***"+textMessage.getText());} }

• 主題(topic)：

-------------生產(chǎn)者------------- @Component public class Topic_Produce {@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate ;@Autowiredprivate Topic topic ;@Scheduled(fixedDelay = 3000)public void produceTopic(){jmsMessagingTemplate.convertAndSend(topic,"主題消息"+ UUID.randomUUID().toString().substring(0,6));} } -------------消費者------------- @Component public class Topic_Consummer {@JmsListener(destination = "${mytopic}")public void receive(TextMessage textMessage) throws Exception{System.out.println("消費者受到訂閱的主題："+textMessage.getText());} }

• 持久化訂閱：

-------------配置Bean------------- /** * 設置持久化訂閱 * 配置文件的方式無法進行配置持久化訂閱。所以需要自己去生成一個持久化訂閱 */ @Component @EnableJms public class ActiveMQConfigBean { @Value("${spring.activemq.broker-url}") private String brokerUrl; @Value("${spring.activemq.user}") private String user; @Value("${spring.activemq.password}") private String password; public ConnectionFactory connectionFactory(){ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(); connectionFactory.setBrokerURL(brokerUrl); connectionFactory.setUserName(user); connectionFactory.setPassword(password); return connectionFactory; } @Bean(name = "jmsListenerContainerFactory") public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() { DefaultJmsListenerContainerFactory defaultJmsListenerContainerFactory = new DefaultJmsListenerContainerFactory(); defaultJmsListenerContainerFactory.setConnectionFactory(connectionFactory()); defaultJmsListenerContainerFactory.setSubscriptionDurable(true); defaultJmsListenerContainerFactory.setClientId("我是持久訂閱者一號"); return defaultJmsListenerContainerFactory; } } -------------消費者------------- @Component public class Topic_Consumer { //需要在監(jiān)聽方法指定連接工廠 @JmsListener(destination = "${myTopicName}",containerFactory = "jmsListenerContainerFactory") public void consumer(TextMessage textMessage) throws JMSException { System.out.println("訂閱著收到消息: " + textMessage.getText()); } }

SpringBoot整合ActiveMQ之Queue與Topoic并存：

• application.properties中定義相關配置項：

spring.jms.pub-sub-domain=true spring.activemq.broker-url=tcp://172.18.1.18:61616 #spring.activemq.user=按實際情況配置 #spring.activemq.password=按實際情況配置 spring.activemq.in-memory=false spring.activemq.pool.enabled=false spring.activemq.pool.maxConnections=2 spring.activemq.pool.expiryTimeout=0 spring.activemq.pool.idleTimeout=30000 spring.activemq.packages.trust-all=true

• 定義配置類：

@Configuration @EnableJms public class JmsConfiguration {// topic模式的ListenerContainer@Beanpublic JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();bean.setPubSubDomain(true);bean.setConnectionFactory(activeMQConnectionFactory);return bean;}// queue模式的ListenerContainer@Beanpublic JmsListenerContainerFactory<?> jmsListenerContainerQueue(ConnectionFactory activeMQConnectionFactory) {DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();bean.setConnectionFactory(activeMQConnectionFactory);return bean;} }

• 定義監(jiān)聽器實現(xiàn)：

@Service public class MQConsumerService {@JmsListener(destination = "portal.admin.topic",containerFactory = "jmsListenerContainerTopic") // 監(jiān)聽指定消息主題public void receiveTopic(String message) {System.out.println(message);}@JmsListener(destination = "portal.admin.queue",containerFactory = "jmsListenerContainerQueue") // 監(jiān)聽指定消息主題public void receiveQueue(String message) {System.out.println(message);} }

ActiveMQ的傳輸協(xié)議

ActiveMQ傳輸協(xié)議簡介：

• ActiveMQ支持的client-broker通訊協(xié)議有：TVP、NIO、UDP、SSL、Http(s)、VM。
• 其中配置Transport Connector的文件在ActiveMQ安裝目錄的conf/activemq.xml中的標簽之內(nèi)。見下圖實際配置：

<transportConnectors><transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/><transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/><transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/><transportConnector name="mqtt" uri="mqtt://0.0.0.0:1884?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/><transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/> </transportConnectors>

• 在上文給出的配置信息中，URI描述信息的頭部都是采用協(xié)議名稱：例如描述
  ①amqp協(xié)議的監(jiān)聽端口時，采用的URI描述格式為“amqp://······”；
  ②描述Stomp協(xié)議的監(jiān)聽端口時，采用URI描述格式為“stomp://······”；
  ③唯獨在進行openwire協(xié)議描述時，URI頭卻采用的“tcp://······”。這是因為ActiveMQ中默認的消息協(xié)議就是openwire

ActiveMQ傳輸協(xié)議有哪些：

• Transmission Control Protocol(TCP)默認：
  ①這是默認的Broker配置，TCP的Client監(jiān)聽端口61616
  ②在網(wǎng)絡傳輸數(shù)據(jù)前，必須要先序列化數(shù)據(jù)，消息是通過一個叫wire protocol的來序列化成字節(jié)流。 ③TCP連接的URI形式如：tcp://HostName:port?key=value&key=value，后面的參數(shù)是可選的。 ④TCP傳輸?shù)牡膬?yōu)點：
  <1>TCP協(xié)議傳輸可靠性高，穩(wěn)定性強
  <2>高效率：字節(jié)流方式傳遞，效率很高
  <3>有效性、可用性：應用廣泛，支持任何平臺
  ⑤關于Transport協(xié)議的可選配置參數(shù)可以參考官網(wǎng)
• New I/O API Protocol(NIO)：
  ①NIO協(xié)議和TCP協(xié)議類似，但NIO更側(cè)重于底層的訪問操作。它允許開發(fā)人員對同一資源可有更多的client調(diào)用和服務器端有更多的負載。
  ②適合使用NIO協(xié)議的場景：
  <1>可能有大量的Client去連接到Broker上，一般情況下，大量的Client去連接Broker是被操作系統(tǒng)的線程所限制的。因此，NIO的實現(xiàn)比TCP需要更少的線程去運行，所以建議使用NIO協(xié)議。
  <2>可能對于Broker有一個很遲鈍的網(wǎng)絡傳輸，NIO比TCP提供更好的性能。
  ③NIO連接的URI形式：nio://hostname:port?key=value&key=value
  ④關于Transport協(xié)議的可選配置參數(shù)可以參考官網(wǎng)
• AMQP協(xié)議：
  ①Advanced Message Queuing Protocol，一個提供統(tǒng)一消息服務的應用層標準高級消息隊列協(xié)議，是應用層協(xié)議的一個開放標準，為面向消息的中間件設計。
  ②基于此協(xié)議的客戶端與消息中間件可傳遞消息，并不受客戶端/中間件不同產(chǎn)品，不同開發(fā)語言等條件限制。
• Stomp協(xié)議：
  ①STOMP，Streaming Text Orientation Message Protocol，是流文本定向消息協(xié)議，是一種為MOM(Message Oriented Middleware，面向消息中間件)設計的簡單文本協(xié)議。
• Secure Sockets Layer Protocol(SSL)：
  ①安全加密協(xié)議。
• MQTT協(xié)議：
  ①MQTT(Message Queuing Telemetry Transport，消息隊列遙測傳輸)是IBM開發(fā)的一個即時通訊協(xié)議，有可能成為物聯(lián)網(wǎng)的重要組成部分。該協(xié)議支持所有平臺，幾乎可以把所有聯(lián)網(wǎng)物品和外部連接起來，被用來當作傳感器和致動器(比如通過Twitter讓房屋聯(lián)網(wǎng))的通信協(xié)議。
• WS協(xié)議(websocket)：
  ①websocket協(xié)議。

配置nio協(xié)議：

• ActiveMQ這些協(xié)議傳輸?shù)牡讓幽J都是使用BIO網(wǎng)絡的IO模型。只有當我們指定使用nio才使用NIO的IO模型。
• 修改配置文件activemq.xml：
  ①在<transportConnectors>節(jié)點下添加如下內(nèi)容：
  <transportConnector name="nio" uri="nio://0.0.0.0:61618?trace=true" />
  ②修改完成后重啟activemq：service activemq restart
  ③查看管理后臺，可以看到頁面多了nio
• NIO協(xié)議增強：
  ①URI格式以"nio"開頭，代表這個端口使用TCP協(xié)議為基礎的NIO網(wǎng)絡模型。但是這樣的設置方式，只能使這個端口支持Openwire協(xié)議。
  ②如果我們既需要使用某一個端口支持NIO網(wǎng)絡模型,又需要它支持多個協(xié)議：
  <1>可以使用auto關鍵字
  <2>使用"+"符號來為端口設置多種特性
  ③配置：<transportConnector name="auto+nio" uri="auto+nio://localhost:5671"/>

ActiveMQ的消息存儲和持久化

ActiveMQ的消息持久化簡介：

• 為了避免意外宕機以后丟失信息，需要做到重啟后可以恢復消息隊列，消息系統(tǒng)一半都會采用持久化機制。ActiveMQ的消息持久化機制有JDBC，AMQ，KahaDB和LevelDB，無論使用哪種持久化方式，消息的存儲邏輯都是一致的。
• 就是在發(fā)送者將消息發(fā)送出去后，消息中心首先將消息存儲到本地數(shù)據(jù)文件、內(nèi)存數(shù)據(jù)庫或者遠程數(shù)據(jù)庫等。再試圖將消息發(fā)給接收者，成功則將消息從存儲中刪除，失敗則繼續(xù)嘗試嘗試發(fā)送。消息中心啟動以后，要先檢查指定的存儲位置是否有未成功發(fā)送的消息，如果有，則會先把存儲位置中的消息發(fā)出去。
• 一句話：ActiveMQ宕機了，消息不會丟失的機制。

ActiveMQ的消息持久化有哪些：

• AMQ Mesage Store(了解）：
  ①AMQ是一種文件存儲形式，它具有寫入速度快和容易恢復的特點。消息存儲再一個個文件中文件的默認大小為32M，當一個文件中的消息已經(jīng)全部被消費，那么這個文件將被標識為可刪除，在下一個清除階段，這個文件被刪除。AMQ適用于ActiveMQ5.3之前的版本。
  ②基于文件的存儲方式，是以前的默認消息存儲，現(xiàn)在不用了。
• KahaDB消息存儲(默認)：
  ①基于日志文件，從ActiveMQ5.4開始默認的持久化插件。
  ②KahaDB是目前默認的存儲方式，可用于任何場景，提高了性能和恢復能力。消息存儲使用一個事務日志和僅僅用一個索引文件來存儲它所有的地址。KahaDB是一個專門針對消息持久化的解決方案，它對典型的消息使用模型進行了優(yōu)化。數(shù)據(jù)被追加到data logs中。當不再需要log文件中的數(shù)據(jù)的時候，log文件會被丟棄。
• JDBC消息存儲：使用JDBC。
• LevelDB消息存儲(了解)：
  ①這種文件系統(tǒng)是從ActiveMQ5.8之后引進的，它和KahaDB非常相似，也是基于文件的本地數(shù)據(jù)庫存儲形式，但是它提供比KahaDB更快的持久性。
  ②但它不使用自定義B-Tree實現(xiàn)來索引獨寫日志，而是使用基于LevelDB的索引。
• JDBC Message Store with ActiveMQ Journal：JDBC加強版。

KahaDB的存儲原理：

• KahaDB在消息保存的目錄中有4類文件和一個lock，跟ActiveMQ的其他幾種文件存儲引擎相比，這就非常簡潔了。
  ①db-number.log：KahaDB存儲消息到預定大小的數(shù)據(jù)紀錄文件中，文件名為db-number.log。當數(shù)據(jù)文件已滿時，一個新的文件會隨之創(chuàng)建，number數(shù)值也會隨之遞增，它隨著消息數(shù)量的增多，如沒32M一個文件，文件名按照數(shù)字進行編號，如db-1.log，db-2.log······。當不再有引用到數(shù)據(jù)文件中的任何消息時，文件會被刪除或者歸檔。
  ②db.data： 該文件包含了持久化的BTree索引，索引了消息數(shù)據(jù)記錄中的消息，它是消息的索引文件，本質(zhì)上是B-Tree（B樹），使用B-Tree作為索引指向db-number。log里面存儲消息。
  ③db.free：當問當前db.data文件里面哪些頁面是空閑的，文件具體內(nèi)容是所有空閑頁的ID
  ④db.redo：用來進行消息恢復，如果KahaDB消息存儲再強制退出后啟動，用于恢復BTree索引。
  ⑤lock：文件鎖，表示當前kahadb獨寫權(quán)限的broker。

<persistenceAdapter><kahaDB directory="${activemq.data}/kahadb"/> </persistenceAdapter>

JDBC存儲消息：

• 添加mysql數(shù)據(jù)庫的驅(qū)動包到lib文件夾
• jdbcPersistenceAdapter配置：
  ①dataSource指定將要引用的持久化數(shù)據(jù)庫的bean名稱。
  ②createTablesOnStartup是否在啟動的時候創(chuàng)建數(shù)據(jù)表，默認值是true,這樣每次啟動都會去創(chuàng)建數(shù)據(jù)表了，一股是第一次啟動的時候設置為true之后改成false。

<persistenceAdapter> <jdbcPersistenceAdapter dataSource="#mysql-ds" createTableOnStartup="true"/> </persistenceAdapter>

• 數(shù)據(jù)庫連接池配置：

<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close"><property name="driverClassName" value="com.mysql.jdbc.Driver"I><property name="url" value="jdbc:mysql/l自己的數(shù)據(jù)庫IP:3306/activemq?relaxAutoCommit=true"> <property name="username" value="自己的數(shù)據(jù)庫用戶名"/><property name="password" value="自己的數(shù)據(jù)庫密碼""/><property name="maxTotal" value="200""/><property name="poolPreparedStatements" value="true"/> </bean>

• 建庫SQL和創(chuàng)表說明：
  ①建一個名為activemq的數(shù)據(jù)庫
  ②如果新建數(shù)據(jù)庫ok，上述配置ok，代碼運行ok，3張表會自動生成
  ③如果表沒生成，可能需要自己創(chuàng)建
  ②三張表的說明：
  <1>ACTIVEMQ_MSGS
  <2>ACTIVEMQ_ACKS
  <3>ACTIVEMQ_LOCK

-------------ACTIVEMQ_MSGS------------- 說明： 消息表，缺省表名為ACTIVEMQ MSGS，queue和topic都存在里面，結(jié)構(gòu)如下數(shù)據(jù)庫字段如下: ID:自增的數(shù)據(jù)庫主鍵 CONTAINER:消息的DestinationMSGID_PROD:消息發(fā)送者的主鍵 MSG_SEQ:是發(fā)送消息的順序，MSGID_PROD+MSG_SEQ可以組成JMS的MessagelDEXPIRATION:消息的過期時間，存儲的是從197O-01-01到現(xiàn)在的毫秒數(shù) MSG:消息本體的Java序列化對象的二進制數(shù)據(jù) PRIORITY:優(yōu)先級，從O-9，數(shù)值越大優(yōu)先級越高-------------ACTIVEMQ_ACKS------------- 說明： activemq_acks用于存儲訂閱關系。如果是持久化Topic，訂閱者和服務器的訂閱關系在這個表保存。 ACTIVEMQ_ACKS表存儲持久訂閱的信息和最后一個持久訂閱接收的消息ID。數(shù)據(jù)庫字段如下: CONTAINER:消息的Destination SUB_DEST:如果是使用Static集群，這個字段會有集群其他系統(tǒng)的信息CLIENT_ID:每個訂閱者都必須有一個唯一的客戶端ID用以區(qū)分 SUB_NAME:訂閱者名稱 SELECTOR:選擇器，可以選擇只消費滿足條件的消息。條件可以用自定義屬性實現(xiàn)，可支持多屬性AND和OR操作LAST_ACKED_ID:記錄消費過的消息的ID。-------------ACTIVEMQ_LOCK------------- 說明： 表activemg_lock在集群環(huán)境中才有用，只有一個Broker可以獲得消息，稱為Master Broker，其他的只能作為備份等 待MasterBroker不可用，才可能成為下一個Master Broker。這個表用于記錄哪個Broker是當前的Master Broker。

• 驗證總結(jié)：
  ①點對點：在點對點類型中當DeliveryMode設置為NON_PERSISTENCE時，消息被保存在內(nèi)存中當DeliveryMode設置為PERSISTENCE時，消息保存在broker的相應的文件或者數(shù)據(jù)庫中。而且點對點類型中消息一旦被Consumer消費，就從數(shù)據(jù)中刪除。消費前的消息會被存放到數(shù)據(jù)庫，上面的消息被消費后被MQ自動刪除。
  ②發(fā)布/訂閱：設置了持久訂閱數(shù)據(jù)庫里面會保存訂閱者的信息，消費者消費所有的數(shù)據(jù)后。ACTIVEMQ_MSGS數(shù)據(jù)表的數(shù)據(jù)并沒有消失。持久化topic的消息不管是否被消費，是否有消費者，產(chǎn)生的數(shù)據(jù)永遠都存在，且只存儲一條。這個是要注意的，持久化的topic大量數(shù)據(jù)后可能導致性能下降。這里就像公總號一樣，消費者消費完后，消息還會保留。
• 小總結(jié)：
  ①如果是queue在沒有消費者消費的情況下會將消息保存到activemq_msgs表中，只要有任意一個消費者消費了，就會刪除。
  ②消費過的消息如果是topic，一般是先啟動消費訂閱者然后再生產(chǎn)的情況下會將持久訂閱者永久保存到qctivemq_acks，而消息則永久保存在activemq_msgs，在acks表中的訂閱者有一個last_ack_id對應了activemq_msgs中的id字段，這樣就知道訂閱者最后收到的消息是哪一條。
• 注意：
  ①在配置關系型數(shù)據(jù)庫作為ActiveMQ的持久化存儲方案時，有坑 數(shù)據(jù)庫jar包注意把對應版本的數(shù)據(jù)庫jar或者你自己使用的非自帶的數(shù)據(jù)庫連接池jar包
  ②createTablesOnStartup屬性默認為true，每次啟動activemq都會自動創(chuàng)建表，在第一次啟動后，應改為false，避免不必要的損失。
  ③java.lang.IllegalStateException: LifecycleProcessor not initialized確認計算機主機名名稱沒有下劃線

JDBC Message store with ActiveMQ Journal：

• 說明：
  ①這種方式克服了JDBC Store的不足，JDBC每次消息過來，都需要去寫庫讀庫。ActiveMQ Journal，使用高速緩存寫入技術(shù)，大大提高了性能。當消費者的速度能夠及時跟上生產(chǎn)者消息的生產(chǎn)速度時，journal文件能夠大大減少需要寫入到DB中的消息。
  ②舉個例子：生產(chǎn)者生產(chǎn)了1000條消息，這1000條消息會保存到journal文件，如果消費者的消費速度很快的情況下，在journal文件還沒有同步到DB之前，消費者已經(jīng)消費了90%的以上消息，那么這個時候只需要同步剩余的10%的消息到DB。如果消費者的速度很慢，這個時候journal文件可以使消息以批量方式寫到DB。
  ③為了高性能，這種方式使用日志文件存儲+數(shù)據(jù)庫存儲。先將消息持久到日志文件，等待一段時間再將未消費的消息持久到數(shù)據(jù)庫。該方式要比JDBC性能要高。
• 配置：

<persistenceFactory><journalPersistenceAdapterFactoryjournalLogFiles="4”journalLogFileSize="32768"useJournal="true"useQuickJournal="true"dataSource="#mysql-ds"dataDirectory="activemq-data"/> </persistenceFactory>

• 總結(jié)：以前是實時寫入mysql，在使用了journal后，數(shù)據(jù)會被journal處理，如果在一定時間內(nèi)journal處理（消費）完了，就不寫入mysql，如果沒消費完，就寫入mysql，起到一個緩存的作用

總結(jié)：

• jdbc效率低，kahaDB效率高，jdbc+Journal效率較高。
• 持久化消息主要指的是：MQ所在服務器宕機了消息不會丟試的機制。
• 持久化機制演變的過程：從最初的AMQ Message Store方案到ActiveMQ V4版本退出的High Performance Journal（高性能事務支持）附件，并且同步推出了關于關系型數(shù)據(jù)庫的存儲方案。ActiveMQ5.3版本又推出了對KahaDB的支持（5.4版本后被作為默認的持久化方案），后來ActiveMQ 5.8版本開始支持LevelDB，到現(xiàn)在5.9提供了標準的Zookeeper+LevelDB集群化方案。
• ActiveMQ消息持久化機制有：

方案原理

AMQ	基于日志文件
KahaDB	基于日志文件，從ActiveMQ5.4開始默認使用
JDBC	基于第三方數(shù)據(jù)庫
Replicated LevelDB Store	從5.9開始提供了LevelDB和Zookeeper的數(shù)據(jù)復制方法，用于Master-slave方式的首選數(shù)據(jù)復制方案。

ActiveMQ多節(jié)點集群

簡介：

• 基于zookeeper和LevelDB搭建ActiveMQ集群。集群僅提供主備方式的高可用集群功能，避免單點故障。
• 引入消息隊列之后該如何保證其高可用性。

三種集群方式對比：

• 基于shareFileSystem共享文件系統(tǒng)（KahaDB）
• 基于JDBC
• 基于可復制的LevelDB

官網(wǎng)集群原理圖：

• 使用Zookeeper集群注冊所有的ActiveMQBroker但只有其中一個Broker可以提供服務，它將被視為Master,其他的Broker處于待機狀態(tài)被視為Slave。如果Master因故障而不能提供服務，Zookeeper會從Slave中選舉出一個Broker充當Master。Slave連接Master并同步他們的存儲狀態(tài)，Slave不接受客戶端連接。所有的存儲操作都將被復制到連接至Maste的Slaves。如果Master宕機得到了最新更新的Slave會變成Master。故障節(jié)點在恢復后會重新加入到集群中并連接Master進入Slave模式。所有需要同步的消息操作都將等待存儲狀態(tài)被復制到其他法定節(jié)點的操作完成才能完成。
• 所以，如給你配置了replicas=3，name法定大小是（3/2）+1 =2。Master將會存儲更新然后等待（2-1）=1個Slave存儲和更新完成，才匯報success，至于為什么是2-1，陽哥的zookeeper講解過自行復習。有一個ode要作為觀察者存在。當一個新的Master被選中，你需要至少保障一個法定mode在線以能夠找到擁有最新狀態(tài)的ode，這個ode才可以成為新的Master。因此，推薦運行至少3個replica nodes以防止一個node失敗后服務中斷。

zookeeper+replicated-leveldb-store的主從集群簡介：

ActiveMQ高級特性

異步投遞Async Sends簡介：

• 對于一個Slow Consumer,使用同步發(fā)送消息可能出現(xiàn)Producer堵塞的情況,慢消費者適合使用異步發(fā)送。
• 同步發(fā)送與異步發(fā)送詳解：
  ①ActiveMQ支持同步，異步兩種發(fā)送的模式將消息發(fā)送到broker，模式的選擇對發(fā)送延時有巨大的影響。producer能達到怎么樣的產(chǎn)出率（產(chǎn)出率=發(fā)送數(shù)據(jù)總量/時間）主要受發(fā)送延時的影響，使用異步發(fā)送可以顯著提高發(fā)送的性能。
  ②ActiveMQ默認使用異步發(fā)送的模式：除非明確指定使用同步發(fā)送的方式或者在未使用事務的前提下發(fā)送持久化的消息，這兩種情況都是同步發(fā)送的。
  ③如果你沒有使用事務且發(fā)送的是持久化的消息，每一次發(fā)送都是同步發(fā)送的且會阻塞producer知道broker返回一個確認，表示消息已經(jīng)被安全的持久化到磁盤。確認機制提供了消息安全的保障，但同時會阻塞客戶端帶來了很大的延時。
  ④很多高性能的應用，允許在失敗的情況下有少量的數(shù)據(jù)丟失。如果你的應用滿足這個特點，你可以使用異步發(fā)送來提高生產(chǎn)率，即使發(fā)送的是持久化的消息。
  ⑤異步發(fā)送它可以最大化producer端的發(fā)送效率。我們通常在發(fā)送消息量比較密集的情況下使用異步發(fā)送，它可以很大的提升Producer性能；不過這也帶來了額外的問題，就是需要消耗更多的Client端內(nèi)存同時也會導致broker端性能消耗增加；此外它不能有效的確保消息的發(fā)送成功。在userAsyncSend=true的情況下客戶端需要容忍消息丟失的可能。
• 異步發(fā)送配置：

public class Jms_TX_Producer {// 方式1。3種方式任選一種private static final String ACTIVEMQ_URL = "tcp://118.24.20.3:61626?jms.useAsyncSend=true";private static final String ACTIVEMQ_QUEUE_NAME = "Async";public static void main(String[] args) throws JMSException {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);// 方式2activeMQConnectionFactory.setUseAsyncSend(true);Connection connection = activeMQConnectionFactory.createConnection();// 方式3((ActiveMQConnection)connection).setUseAsyncSend(true);connection.start();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);MessageProducer producer = session.createProducer(queue);try {for (int i = 0; i < 3; i++) {TextMessage textMessage = session.createTextMessage("tx msg--" + i);producer.send(textMessage);}System.out.println("消息發(fā)送完成");} catch (Exception e) {e.printStackTrace();} finally {producer.close();session.close();connection.close();}} }

• 異步消息如何確定發(fā)送成功?
  ①異步發(fā)送丟失消息的場景是：生產(chǎn)者設置userAsyncSend=true，使用producer.send(msg)持續(xù)發(fā)送消息。如果消息不阻塞，生產(chǎn)者會認為所有send的消息均被成功發(fā)送至MQ。如果MQ突然宕機，此時生產(chǎn)者端內(nèi)存中尚未被發(fā)送至MQ的消息都會丟失。
  ②所以正確的異步發(fā)送方法是需要接收回調(diào)的。同步發(fā)送和異步發(fā)送的區(qū)別就在此，同步發(fā)送等send不阻塞了就表示一定發(fā)送成功了，異步發(fā)送需要客戶端回執(zhí)并由客戶端再判斷一次是否發(fā)送成功。

public class Jms_TX_Producer {private static final String ACTIVEMQ_URL = "tcp://118.24.20.3:61626";private static final String ACTIVEMQ_QUEUE_NAME = "Async";public static void main(String[] args) throws JMSException {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);activeMQConnectionFactory.setUseAsyncSend(true);Connection connection = activeMQConnectionFactory.createConnection();connection.start();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);ActiveMQMessageProducer activeMQMessageProducer = (ActiveMQMessageProducer)session.createProducer(queue);try {for (int i = 0; i < 3; i++) {TextMessage textMessage = session.createTextMessage("tx msg--" + i);textMessage.setJMSMessageID(UUID.randomUUID().toString()+"orderAtguigu");final String msgId = textMessage.getJMSMessageID();activeMQMessageProducer.send(textMessage, new AsyncCallback() {public void onSuccess() {System.out.println("成功發(fā)送消息Id:"+msgId);}public void onException(JMSException e) {System.out.println("失敗發(fā)送消息Id:"+msgId);}});}System.out.println("消息發(fā)送完成");} catch (Exception e) {e.printStackTrace();} finally {activeMQMessageProducer.close();session.close();connection.close();}} }

延遲投遞和定時投遞簡介：

• 四大屬性：

Property nametypedescription

AMQ_SCHEDULED_DELAY	long	延遲投遞的時間
AMQ_SCHEDULED_PERIOD	long	重復投遞的時間間隔
AMQ_SCHEDULED_REPEAT	int	重復投遞次數(shù)
AMQ_SCHEDULED_CRON	string	Cron表達式

• 配置：要在activemq.xml中配置schedulerSupport屬性為true

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost"dataDirectory="${activemq.data}" schedulerSupport="true" >

• Java代碼里面封裝的輔助消息類型：ScheduledMessage

public class Jms_TX_Producer {private static final String ACTIVEMQ_URL = "tcp://118.24.20.3:61626";private static final String ACTIVEMQ_QUEUE_NAME = "Schedule01";public static void main(String[] args) throws JMSException {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);Connection connection = activeMQConnectionFactory.createConnection();connection.start();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);MessageProducer messageProducer = session.createProducer(queue);long delay = 10*1000;long period = 5*1000;int repeat = 3 ;try {for (int i = 0; i < 3; i++) {TextMessage textMessage = session.createTextMessage("tx msg--" + i);// 延遲的時間textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);// 重復投遞的時間間隔textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);// 重復投遞的次數(shù)textMessage.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);// 此處的意思：該條消息，等待10秒，之后每5秒發(fā)送一次，重復發(fā)送3次。messageProducer.send(textMessage);}System.out.println("消息發(fā)送完成");} catch (Exception e) {e.printStackTrace();} finally {messageProducer.close();session.close();connection.close();}} }

消息消費的重試機制：

• 是什么？
  ①消費者收到消息，之后出現(xiàn)異常了，沒有告訴broker確認收到該消息，broker會嘗試再將該消息發(fā)送給消費者。嘗試n次，如果消費者還是沒有確認收到該消息，那么該消息將被放到死信隊列重，之后broker不會再將該消息發(fā)送給消費者。
• 具體哪些情況會引發(fā)消息重發(fā)？
  ①Client用了transactions且再session中調(diào)用了rollback
  ②Client用了transactions且再調(diào)用commit之前關閉或者沒有commit
  ③Client再CLIENT_ACKNOWLEDGE的傳遞模式下，session中調(diào)用了recover
• 請說說消息重發(fā)時間間隔和重發(fā)次數(shù)？
  ①間隔：1
  ②次數(shù)：6
• 有毒消息Poison ACK：
  ①一個消息被redelivedred超過默認的最大重發(fā)次數(shù)（默認6次）時，消費的回個MQ發(fā)一個“poison ack”表示這個消息有毒，告訴broker不要再發(fā)了。這個時候broker會把這個消息放到DLQ（私信隊列）。
• 屬性說明：
  
• 修改配置參數(shù)：

public class Jms_TX_Consumer {private static final String ACTIVEMQ_URL = "tcp://118.24.20.3:61626";private static final String ACTIVEMQ_QUEUE_NAME = "dead01";public static void main(String[] args) throws JMSException, IOException {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);// 修改默認參數(shù)，設置消息消費重試3次RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();redeliveryPolicy.setMaximumRedeliveries(3);activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);Connection connection = activeMQConnectionFactory.createConnection();connection.start();final Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);MessageConsumer messageConsumer = session.createConsumer(queue);messageConsumer.setMessageListener(new MessageListener() {public void onMessage(Message message) {if (message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;try {System.out.println("***消費者接收到的消息: " + textMessage.getText());//session.commit();}catch (Exception e){e.printStackTrace();}}}});System.in.read();messageConsumer.close();session.close();connection.close();} }

• 整合spring：
  

死信隊列：

• 簡介：
  ①異常消息規(guī)避處理的集合，主要處理失敗的消息。
  ②ActiveMQ中引入了“死倍隊列”(Dead Letter Queue〉的概念。即一條消息再被重發(fā)了多次后（默認為重發(fā)6次redeliveryCounter==6)，將會被ActiveMQ移入“死信隊列”。開發(fā)人員可以在這個Queue中查看處理出錯的消息，進行人工干預。
• 死信隊列控制臺：
• 使用：
  
• 死信隊列的配置（一般采用默認）：
  ①sharedDeadLetterStrategy：
  <1>不管是queue還是topic，失敗的消息都放到這個隊列中。下面修改activemq.xml的配置，可以達到修改隊列的名字。
  <2>將所有的eadLetter保存在一個共享的隊列中，這是ActiveMQ broker端默認的策略。共享隊列默認為“ActiveMQ.DLQ”，可以通過“deadLetterQueue”屬性來設定。
  ②individualDeadLetterStrategy：
  <1>可以為queue和topic單獨指定兩個死信隊列。還可以為某個話題，單獨指定一個死信隊列。
  ③自動刪除過期消息：
  <1>過期消息是值生產(chǎn)者指定的過期時間，超過這個時間的消息。
  <2>有時需要直接刪除過期的消息而不需要發(fā)送到死隊列中，“processExpired”表示是否將過期消息放入死信隊列，默認為true。
  ④存放非持久消息到死信隊列中：
  <1>默認情況下，Activemq不會把非持久的死消息發(fā)送到死信隊列中。
  <2>processNonPersistent”表示是否將“非持久化”消息放入死信隊列，默認為false。
  <3>非持久性如果你想把非持久的消息發(fā)送到死隊列中，需要設置屬性processNonPersistent=“true"

消息不被重復消費，冪等性：

• 網(wǎng)絡延遲傳輸中，會造成進行MQ重試中，在重試過程中，可能會造成重復消費。
• 如果消息是做數(shù)據(jù)庫的插入操作，給這個消息做一個唯一主鍵，那么就算出現(xiàn)重復消費的情況，就會導致主鍵沖突，避免數(shù)據(jù)庫出現(xiàn)臟數(shù)據(jù)。
• 如果上面兩種情況還不行，準備一個第三服務方來做消費記錄。以redis為例，給消息分配一個全局id，只要消費過該消息，將<id,message>以K-V形式寫入redis。那消費者開始消費前，先去redis中查詢有沒消費記錄即可。
• 冪等性如何解決，根據(jù)messageid去查這個消息是否被消費了。

總結(jié)

以上是生活随笔為你收集整理的ActiveMQ知识概括的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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