ActiveMQ知識概括 ActiveMQ簡介 Java實現ActiveMQ JMS規范與落地 ActiveMQ的broker Spring，SpringBoot整合ActiveMQ ActiveMQ的傳輸協議 ActiveMQ的消息存儲和持久化 ActiveMQ多節點集群 ActiveMQ高級特性
ActiveMQ簡介
ActiveMQ安裝：
安裝步驟： 啟動時指定日志輸出文件： 查看程序啟動是否成功的3種方式（通用）： ActiveMQ控制臺：
訪問activemq管理頁面地址：http://IP地址:8161/。默認的用戶名和密碼是admin/admin。 備注： 默認程序連接activemq(JMS服務)是不需要密碼的，為了安裝起見，一般都會設置密碼，提高安全性。 ActiveMQ控制臺之隊列： ActiveMQ控制臺之主題： ActiveMQ控制臺之訂閱者： Java實現ActiveMQ
pom.xml導入依賴：
< dependency> < groupId> org.apache.activemq
</ groupId> < artifactId> activemq-all
</ artifactId> < version> 5.15.9
</ version> </ dependency> < dependency> < groupId> org.apache.xbean
</ groupId> < artifactId> xbean-spring
</ artifactId> < version> 3.16
</ version> </ dependency> JMS編碼總體規范：
架構： JMS開發的基本步驟： Destination簡介：
Destination是目的地。下面拿jvm和mq，做個對比。目的地，我們可以理解為是數據存儲的地方。 Destination分為兩種：隊列和主題。 隊列消息（Queue）總結：
兩種消費方式： 隊列的特點： 消息消費情況： 主題消息(Topic)介紹：
在發布訂閱消息傳遞域中，目的地被稱為主題（topic） 發布/訂閱消息傳遞域的特點如下： tpoic和queue對比：
比較項目Topic模式隊列Queue模式隊列 工作模式. “訂閱-發布"模式，如果當前沒有訂閱者，消息將會被丟棄。如果有多個訂閱者，那么這些訂閱者都會收到消息 “負載均衡"模式，如果當前沒有消費者，消息也不會云棄;如果有多個消費者，那么—條消息也只會發送始其中一個消費者，并且要求消費者ack信息 有無狀態 無狀態 Queue數據默認會在mq服務器上以文件形式保存，比如Active MQ—般保存在SAMQ_HOME\datakr-storeldata下面。也可以配置成DB存儲。 傳遞完整性 如果沒有訂閱者，消息會被丟棄 消息不會云棄 處理效率 由于消息要按照訂閱者的數量進行復制，所以處理性能會隨著訂閱者的增加而明顯降低,并且還要結合不同消息協議自身的性能差異 由于—條消息只發送給—個消費者，所以就算消費者再多，性能也不會有明顯降低。當然不同消息協議的具體性能也是有差異的
JMS規范與落地
JMS是什么：
JMS是Java消息服務 Java消息服務指的是兩個應用程序之間進行異步通信的API，它為標準協議和消息服務提供了一組通用接口，包括創建、發送、讀取消息等，用于支持Java應用程序開發。在JavaEE中，當兩個應用程序使用JMS進行通信時，它們之間不是直接相連的，而是通過一個共同的消息收發服務組件關聯起來以達到解耦/異步削峰的效果。 JMS的組成結構和特點：
消息頭：
JMS的消息頭有哪些屬性： 說明： 消息的生產者可以set這些屬性，消息的消費者可以get這些屬性。這些屬性在send方法里面也可以設置。 消息體：
封裝具體的消息數據 5種消息體格式： 發送和接受的消息體類型必須一致對應 消息屬性：
如果需要除消息頭字段之外的值，那么可以使用消息屬性。他是識別/去重/重點標注等操作，非常有用的方法。 他們是以屬性名和屬性值對的形式制定的。可以將屬性是為消息頭得擴展，屬性指定一些消息頭沒有包括的附加信息，比如可以在屬性里指定消息選擇器。消息的屬性就像可以分配給一條消息的附加消息頭一樣。它們允許開發者添加有關消息的不透明附加信息。它們還用于暴露消息選擇器在消息過濾時使用的數據。 下圖是設置消息屬性的API：set對應類型Property(String name，對應類型 value) JMS的可靠性：
PERSISTENT：持久性 Transaction：事務 Acknowledge：簽收 消息的持久化：
什么是持久化消息？ 參數設置說明： 持久的Queue：持久化消息這是隊列的默認傳遞模式，此模式保證這些消息只被傳送一次和成功使用一次。對于這些消息，可靠性是優先考慮的因素。可靠性的另一個重要方面是確保持久性消息傳送至目標后，消息服務在向消費者傳送它們之前不會丟失這些消息。 持久的Topic：一定要先運行一次消費者,類似于像MQ注冊,我訂閱了這個主題。然后再運行主題生產者，無論消費著是否在線,都會接收到,在線的立即接收到,不在線的等下次上線把沒接收到的接收。類似微信公眾號訂閱發布。 消息事務：
producer提交時的事務： consumer消費時的事務： 事務偏生產者/簽收偏消費者！ 消息簽收：
非事務： 事務： 簽收和事務的關系： JMS的點對點總結：
點對點模型是基于隊列的，生產者發送消息到隊列，消費者從隊列接收消息，隊列的存在使得消息的異步傳輸成為可能。和我們平時給朋友發送短信類似。 JMS的發布訂閱總結：
非持久訂閱： 持久訂閱： 用哪個? ActiveMQ的broker
簡介：
相當于一個ActiveMQ服務器實例說白了，Broker其實就是實現了用代碼的形式啟動ActiveMQ將MQ嵌入到Java代碼中，以便隨時用隨時啟動，在用的時候再去啟動這樣能節省了資源，也保證了可用性。 嵌入式Broker：
< 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的包： < dependency> < groupId> org.apache.activemq
</ groupId> < artifactId> activemq-all
</ artifactId> < version> 5.10.0
</ version> </ dependency> < dependency> < groupId> com.fasterxml.jackson.core
</ groupId> < artifactId> jackson-databind
</ artifactId> < version> 2.10.1
</ version> </ dependency> < dependency> < groupId> org.apache.activemq
</ groupId> < artifactId> activemq-pool
</ artifactId> < version> 5.15.10
</ version> </ dependency> < dependency> < groupId> org.springframework
</ groupId> < artifactId> spring-jms
</ artifactId> < version> 5.2.1.RELEASE
</ version> </ dependency> < dependency> < groupId> org.apache.xbean
</ groupId> < artifactId> xbean-spring
</ artifactId> < version> 4.15
</ version> </ dependency> <?xml version="1.0" encoding="UTF-8"?>
< beansxmlns = " 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-scanbase-package = " com.activemq.demo" /> < beanid = " connectionFactory" class = " org.apache.activemq.pool.PooledConnectionFactory" destroy-method = " stop" > < propertyname = " connectionFactory" > < beanclass = " org.apache.activemq.spring.ActiveMQConnectionFactory" > < propertyname = " brokerURL" value = " tcp://192.168.10.130:61616" /> </ bean> </ property> < propertyname = " maxConnections" value = " 100" /> </ bean> < beanid = " destinationQueue" class = " org.apache.activemq.command.ActiveMQQueue" > < constructor-argindex = " 0" value = " spring-active-queue" /> </ bean> < beanid = " destinationTopic" class = " org.apache.activemq.command.ActiveMQTopic" > < constructor-argindex = " 0" value = " spring-active-topic" /> </ bean> < beanid = " jmsTemplate" class = " org.springframework.jms.core.JmsTemplate" > < propertyname = " connectionFactory" ref = " connectionFactory" /> < propertyname = " defaultDestination" ref = " destinationQueue" /> < propertyname = " messageConverter" > < beanclass = " org.springframework.jms.support.converter.SimpleMessageConverter" /> </ property> </ bean> </ beans>
---------------生產者------------------
@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); }
}
---------------生產者------------------
@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); //直接調用application.xml里面創建的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); //直接調用application.xml里面創建的destinationTopic這個bean設置為目的地就行了 springMQConsumer.jmsTemplate.setDefaultDestination(((Destination) applicationContext.getBean("destinationTopic"))); String returnValue = (String) springMQConsumer.jmsTemplate.receiveAndConvert(); System.out.println("****消費者收到的消息: " + returnValue); }
}
在Spring里面實現消費者不啟動，直接通過配置監聽完成： < beanid = " jmscontainer" class = " org.springframework.jms.1listener.DefaultlessageListenerContainer" > < propertyname = " connectionFactory" ref = " jmsFactory" /> < propertyname = " destination" ref = " destinationTopic" /> < propertyname = " messageListener" ref = " myMessageListener" /> </ bean>
//實現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：
< dependency> < groupId> org.springframework.boot
</ groupId> < artifactId> spring-boot-starter-activemq
</ artifactId> < version> 2.1.5.RELEASE
</ version> </ dependency>
# 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
@Component
@EnableJms
//開啟Springboot的Jms
public class ConfigBean { @Value("myQueueName") private String myQueueName; @Bean public ActiveMQQueue queue() { //創建一個ActiveMQQueue return new ActiveMQQueue(myQueueName); }@Value("${myTopicName}") private String topicName; @Bean public ActiveMQTopic activeMQTopic() { //創建一個ActiveMQTopicreturn new ActiveMQTopic(topicName); }
}
-------------生產者-------------
@Component
public class Queue_Produce {// JMS模板@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate ;// 這個是我們配置的隊列目的地@Autowiredprivate Queue queue ;// 發送消息public void produceMessage(){// 一參是目的地，二參是消息的內容jmsMessagingTemplate.convertAndSend(queue,"****"+ UUID.randomUUID().toString().substring(0,6));}// 定時任務。每3秒執行一次。非必須代碼，僅為演示。@Scheduled(fixedDelay = 3000)public void produceMessageScheduled(){produceMessage();}
}
-------------消費者-------------
@Component
public class Queue_consummer {// 注冊一個監聽器。destination指定監聽的主題。@JmsListener(destination = "${myqueue}")public void receive(TextMessage textMessage) throws Exception{System.out.println(" *** 消費者收到消息 ***"+textMessage.getText());}
}
-------------生產者-------------
@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 { //需要在監聽方法指定連接工廠 @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 { @Bean public JmsListenerContainerFactory < ? > jmsListenerContainerTopic ( ConnectionFactory activeMQConnectionFactory
) { DefaultJmsListenerContainerFactory bean
= new DefaultJmsListenerContainerFactory ( ) ; bean
. setPubSubDomain ( true ) ; bean
. setConnectionFactory ( activeMQConnectionFactory
) ; return bean
; } @Bean public JmsListenerContainerFactory < ? > jmsListenerContainerQueue ( ConnectionFactory activeMQConnectionFactory
) { DefaultJmsListenerContainerFactory bean
= new DefaultJmsListenerContainerFactory ( ) ; bean
. setConnectionFactory ( activeMQConnectionFactory
) ; return bean
; }
}
@Service
public class MQConsumerService {@JmsListener(destination = "portal.admin.topic",containerFactory = "jmsListenerContainerTopic") // 監聽指定消息主題public void receiveTopic(String message) {System.out.println(message);}@JmsListener(destination = "portal.admin.queue",containerFactory = "jmsListenerContainerQueue") // 監聽指定消息主題public void receiveQueue(String message) {System.out.println(message);}
}
ActiveMQ的傳輸協議
ActiveMQ傳輸協議簡介：
ActiveMQ支持的client-broker通訊協議有：TVP、NIO、UDP、SSL、Http(s)、VM。 其中配置Transport Connector的文件在ActiveMQ安裝目錄的conf/activemq.xml中的標簽之內。見下圖實際配置： < transportConnectors> < transportConnectorname = " openwire" uri = " tcp://0.0.0.0:61616?maximumConnections=1000& wireFormat.maxFrameSize=104857600" /> < transportConnectorname = " amqp" uri = " amqp://0.0.0.0:5672?maximumConnections=1000& wireFormat.maxFrameSize=104857600" /> < transportConnectorname = " stomp" uri = " stomp://0.0.0.0:61613?maximumConnections=1000& wireFormat.maxFrameSize=104857600" /> < transportConnectorname = " mqtt" uri = " mqtt://0.0.0.0:1884?maximumConnections=1000& wireFormat.maxFrameSize=104857600" /> < transportConnectorname = " ws" uri = " ws://0.0.0.0:61614?maximumConnections=1000& wireFormat.maxFrameSize=104857600" /> </ transportConnectors> 在上文給出的配置信息中，URI描述信息的頭部都是采用協議名稱：例如描述 ActiveMQ傳輸協議有哪些：
Transmission Control Protocol(TCP)默認： New I/O API Protocol(NIO)： AMQP協議： Stomp協議： Secure Sockets Layer Protocol(SSL)： MQTT協議： WS協議(websocket)： 配置nio協議：
ActiveMQ這些協議傳輸的底層默認都是使用BIO網絡的IO模型。只有當我們指定使用nio才使用NIO的IO模型。 修改配置文件activemq.xml： NIO協議增強： ActiveMQ的消息存儲和持久化
ActiveMQ的消息持久化簡介：
為了避免意外宕機以后丟失信息，需要做到重啟后可以恢復消息隊列，消息系統一半都會采用持久化機制。ActiveMQ的消息持久化機制有JDBC，AMQ，KahaDB和LevelDB，無論使用哪種持久化方式，消息的存儲邏輯都是一致的。 就是在發送者將消息發送出去后，消息中心首先將消息存儲到本地數據文件、內存數據庫或者遠程數據庫等。再試圖將消息發給接收者，成功則將消息從存儲中刪除，失敗則繼續嘗試嘗試發送。消息中心啟動以后，要先檢查指定的存儲位置是否有未成功發送的消息，如果有，則會先把存儲位置中的消息發出去。 一句話：ActiveMQ宕機了，消息不會丟失的機制。 ActiveMQ的消息持久化有哪些：
AMQ Mesage Store(了解）： KahaDB消息存儲(默認)： JDBC消息存儲：使用JDBC。 LevelDB消息存儲(了解)： JDBC Message Store with ActiveMQ Journal：JDBC加強版。 KahaDB的存儲原理：
KahaDB在消息保存的目錄中有4類文件和一個lock，跟ActiveMQ的其他幾種文件存儲引擎相比，這就非常簡潔了。 < persistenceAdapter> < kahaDBdirectory = " ${activemq.data}/kahadb" /> </ persistenceAdapter> JDBC存儲消息：
添加mysql數據庫的驅動包到lib文件夾 jdbcPersistenceAdapter配置： < persistenceAdapter> < jdbcPersistenceAdapterdataSource = " #mysql-ds" createTableOnStartup = " true" /> </ persistenceAdapter> < beanid = " mysql-ds" class = " org.apache.commons.dbcp2.BasicDataSource" destroy-method = " close" > < propertyname = " driverClassName" value = " com.mysql.jdbc.Driver" I > < propertyname = " url" value = " jdbc:mysql/l自己的數據庫IP:3306/activemq?relaxAutoCommit=true" > < propertyname = " username" value = " 自己的數據庫用戶名" /> < propertyname = " password" value = " 自己的數據庫密碼" " /> < propertyname = " maxTotal" value = " 200" " /> < propertyname = " poolPreparedStatements" value = " true" /> </ bean> 建庫SQL和創表說明：
-------------ACTIVEMQ_MSGS-------------
說明：
消息表，缺省表名為ACTIVEMQ MSGS，queue和topic都存在里面，結構如下數據庫字段如下:
ID:自增的數據庫主鍵
CONTAINER:消息的DestinationMSGID_PROD:消息發送者的主鍵
MSG_SEQ:是發送消息的順序，MSGID_PROD+MSG_SEQ可以組成JMS的MessagelDEXPIRATION:消息的過期時間，存儲的是從197O-01-01到現在的毫秒數
MSG:消息本體的Java序列化對象的二進制數據
PRIORITY:優先級，從O-9，數值越大優先級越高-------------ACTIVEMQ_ACKS-------------
說明：
activemq_acks用于存儲訂閱關系。如果是持久化Topic，訂閱者和服務器的訂閱關系在這個表保存。
ACTIVEMQ_ACKS表存儲持久訂閱的信息和最后一個持久訂閱接收的消息ID。數據庫字段如下:
CONTAINER:消息的Destination
SUB_DEST:如果是使用Static集群，這個字段會有集群其他系統的信息CLIENT_ID:每個訂閱者都必須有一個唯一的客戶端ID用以區分
SUB_NAME:訂閱者名稱
SELECTOR:選擇器，可以選擇只消費滿足條件的消息。條件可以用自定義屬性實現，可支持多屬性AND和OR操作LAST_ACKED_ID:記錄消費過的消息的ID。-------------ACTIVEMQ_LOCK-------------
說明：
表activemg_lock在集群環境中才有用，只有一個Broker可以獲得消息，稱為Master Broker，其他的只能作為備份等
待MasterBroker不可用，才可能成為下一個Master Broker。這個表用于記錄哪個Broker是當前的Master Broker。
驗證總結： 小總結： 注意： JDBC Message store with ActiveMQ Journal：
說明： 配置： < persistenceFactory> < journalPersistenceAdapterFactoryjournalLogFiles = " 4”journalLogFileSize=" 32768" useJournal = " true" useQuickJournal = " true" dataSource = " #mysql-ds" dataDirectory = " activemq-data" /> </ persistenceFactory> 總結：以前是實時寫入mysql，在使用了journal后，數據會被journal處理，如果在一定時間內journal處理（消費）完了，就不寫入mysql，如果沒消費完，就寫入mysql，起到一個緩存的作用 總結：
jdbc效率低，kahaDB效率高，jdbc+Journal效率較高。 持久化消息主要指的是：MQ所在服務器宕機了消息不會丟試的機制。 持久化機制演變的過程：從最初的AMQ Message Store方案到ActiveMQ V4版本退出的High Performance Journal（高性能事務支持）附件，并且同步推出了關于關系型數據庫的存儲方案。ActiveMQ5.3版本又推出了對KahaDB的支持（5.4版本后被作為默認的持久化方案），后來ActiveMQ 5.8版本開始支持LevelDB，到現在5.9提供了標準的Zookeeper+LevelDB集群化方案。 ActiveMQ消息持久化機制有： 方案原理 AMQ 基于日志文件 KahaDB 基于日志文件，從ActiveMQ5.4開始默認使用 JDBC 基于第三方數據庫 Replicated LevelDB Store 從5.9開始提供了LevelDB和Zookeeper的數據復制方法，用于Master-slave方式的首選數據復制方案。
ActiveMQ多節點集群
簡介：
基于zookeeper和LevelDB搭建ActiveMQ集群。集群僅提供主備方式的高可用集群功能，避免單點故障。 引入消息隊列之后該如何保證其高可用性。 三種集群方式對比：
基于shareFileSystem共享文件系統（KahaDB） 基于JDBC 基于可復制的LevelDB 官網集群原理圖：
使用Zookeeper集群注冊所有的ActiveMQBroker但只有其中一個Broker可以提供服務，它將被視為Master,其他的Broker處于待機狀態被視為Slave。如果Master因故障而不能提供服務，Zookeeper會從Slave中選舉出一個Broker充當Master。Slave連接Master并同步他們的存儲狀態，Slave不接受客戶端連接。所有的存儲操作都將被復制到連接至Maste的Slaves。如果Master宕機得到了最新更新的Slave會變成Master。故障節點在恢復后會重新加入到集群中并連接Master進入Slave模式。所有需要同步的消息操作都將等待存儲狀態被復制到其他法定節點的操作完成才能完成。 所以，如給你配置了replicas=3，name法定大小是（3/2）+1 =2。Master將會存儲更新然后等待（2-1）=1個Slave存儲和更新完成，才匯報success，至于為什么是2-1，陽哥的zookeeper講解過自行復習。有一個ode要作為觀察者存在。當一個新的Master被選中，你需要至少保障一個法定mode在線以能夠找到擁有最新狀態的ode，這個ode才可以成為新的Master。因此，推薦運行至少3個replica nodes以防止一個node失敗后服務中斷。 zookeeper+replicated-leveldb-store的主從集群簡介：
ActiveMQ高級特性
異步投遞Async Sends簡介：
對于一個Slow Consumer,使用同步發送消息可能出現Producer堵塞的情況,慢消費者適合使用異步發送。 同步發送與異步發送詳解： 異步發送配置：
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("消息發送完成");} catch (Exception e) {e.printStackTrace();} finally {producer.close();session.close();connection.close();}}
}
異步消息如何確定發送成功?
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("成功發送消息Id:"+msgId);}public void onException(JMSException e) {System.out.println("失敗發送消息Id:"+msgId);}});}System.out.println("消息發送完成");} 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 重復投遞次數 AMQ_SCHEDULED_CRON string Cron表達式
配置：要在activemq.xml中配置schedulerSupport屬性為true < brokerxmlns = " 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);// 重復投遞的次數textMessage.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);// 此處的意思：該條消息，等待10秒，之后每5秒發送一次，重復發送3次。messageProducer.send(textMessage);}System.out.println("消息發送完成");} catch (Exception e) {e.printStackTrace();} finally {messageProducer.close();session.close();connection.close();}}
}
消息消費的重試機制：
是什么？ 具體哪些情況會引發消息重發？ 請說說消息重發時間間隔和重發次數？ 有毒消息Poison ACK： 屬性說明： 修改配置參數：
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);// 修改默認參數，設置消息消費重試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();}
}
死信隊列：
簡介： 死信隊列控制臺： 使用： 死信隊列的配置（一般采用默認）： 消息不被重復消費，冪等性：
網絡延遲傳輸中，會造成進行MQ重試中，在重試過程中，可能會造成重復消費。 如果消息是做數據庫的插入操作，給這個消息做一個唯一主鍵，那么就算出現重復消費的情況，就會導致主鍵沖突，避免數據庫出現臟數據。 如果上面兩種情況還不行，準備一個第三服務方來做消費記錄。以redis為例，給消息分配一個全局id，只要消費過該消息，將<id,message>以K-V形式寫入redis。那消費者開始消費前，先去redis中查詢有沒消費記錄即可。 冪等性如何解決，根據messageid去查這個消息是否被消費了。 總結
以上是生活随笔 為你收集整理的ActiveMQ知识概括 的全部內容，希望文章能夠幫你解決所遇到的問題。
如果覺得生活随笔 網站內容還不錯，歡迎將生活随笔 推薦給好友。
