ActiveMQ 集群搭建與高可用方案設計（二）

五、高可用方案設計與優化

（一）Zookeeper 在 ActiveMQ 集群中的應用

  1. 作用：在 ActiveMQ 集群中，Zookeeper 扮演著至關重要的角色。它主要用于選舉 Master 節點，通過其內部的選舉機制，從眾多的 ActiveMQ Broker 節點中挑選出一個作為 Master，其他節點則作為 Slave。只有 Master 節點能夠對外提供服務，當 Master 節點因為故障不能正常工作時，Zookeeper 會利用選舉機制，迅速從 Slave 節點中選舉出一個新的 Master 節點，繼續對外提供消息服務，從而有效排除單點故障引起的服務中斷，保證了集群的高可用性 。

同時，Zookeeper 還負責維護集群的一致性。它通過分布式協調服務，幫助 ActiveMQ 集群中的各個 Broker 節點保持數據同步。各個 Broker 節點會將一些關鍵信息，如消息隊列的狀態、消費者的訂閱信息等，存儲在 Zookeeper 的節點上。當某個 Broker 節點發生變化時，Zookeeper 會及時通知其他節點，使它們能夠同步更新狀態，確保整個集群的數據一致性 。

  1. Zookeeper 集群搭建步驟：
    • 下載與解壓：從 Zookeeper 官方網站（Apache ZooKeeper ）下載合適版本的安裝包，如apache-zookeeper-3.8.0-bin.tar.gz。下載完成后，將安裝包上傳到服務器，使用命令tar -zxvf apache-zookeeper-3.8.0-bin.tar.gz解壓到指定目錄，例如/usr/local/zookeeper 。
    • 配置文件修改：進入/usr/local/zookeeper/conf目錄，將zoo_sample.cfg文件復制一份并改名為zoo.cfg。編輯zoo.cfg文件，修改以下關鍵配置：
 

tickTime=2000 # 基本時間單元，以毫秒為單位，用于心跳檢測等
 

initLimit=10 # Leader和Follower之間初始連接時能容忍的最多心跳數
 

syncLimit=5 # Leader和Follower之間請求和應答之間能容忍的最多心跳數
 

dataDir=/usr/local/zookeeper/data # 數據存儲目錄，需要提前創建該目錄
 

clientPort=2181 # 客戶端連接端口
 

server.1=192.168.1.100:2888:3888 # 配置集群中的服務器，格式為server.id=ip:port1:port2，id為服務器的唯一標識，port1用于Leader和Follower之間的通信，port2用于選舉
 

server.2=192.168.1.101:2888:3888
 

server.3=192.168.1.102:2888:3888
 

  • 創建 myid 文件：在dataDir指定的目錄（/usr/local/zookeeper/data）下創建myid文件，文件內容為當前服務器的id值。例如，在192.168.1.100服務器上，myid文件內容為1；在192.168.1.101服務器上，myid文件內容為2；在192.168.1.102服務器上，myid文件內容為3 。
 

  • 啟動 Zookeeper 集群：分別在三臺服務器上進入/usr/local/zookeeper/bin目錄，執行./zkServer.sh start命令啟動 Zookeeper 服務。可以使用./zkServer.sh status命令查看服務狀態，正常情況下，會有一個節點被選舉為 Leader，其他節點為 Follower 。
 

  1. ActiveMQ 與 Zookeeper 集成配置：在 ActiveMQ 的activemq.xml配置文件中，修改persistenceAdapter元素，使用replicatedLevelDB實現基于 Zookeeper 和 LevelDB 的持久化和集群管理。配置示例如下：
 

 

<persistenceAdapter>
 

<replicatedLevelDB
 

directory="${activemq.data}/leveldb"
 

replicas="3"
 

bind="tcp://0.0.0.0:62222"
 

zkAddress="192.168.1.100:2181,192.168.1.101:2181,192.168.1.102:2181"
 

hostname="192.168.1.100"
 

zkPath="/activemq/leveldb-stores/"/>
 

</persistenceAdapter>
 

其中，directory指定 LevelDB 數據存儲目錄；replicas表示集群中的節點數量，一般設置為奇數；bind指定集群內部通信的端口；zkAddress指定 Zookeeper 集群的地址；hostname指定當前節點的 IP 地址；zkPath指定 ActiveMQ 在 Zookeeper 中存儲選舉信息和數據同步的路徑 。
 

（二）消息持久化策略
 

  1. KahaDB：KahaDB 是 ActiveMQ 從 5.4 版本開始的默認持久化方式，它基于日志文件進行消息存儲 。在activemq.xml文件中，默認配置如下：
 

 

<persistenceAdapter>
 

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

</persistenceAdapter>
 

KahaDB 使用一個事務日志和一個索引文件來存儲所有的消息地址。消息存儲在db-n.log文件中，當文件滿了會創建新的文件，n會遞增。db.data文件是 BTree 索引文件，用于索引消息數據記錄中的消息；db.free文件記錄當前db.data文件中哪些頁面是空閑的；db.redo文件用于在 KahaDB 消息存儲強制退出后啟動時恢復 BTree 索引；lock文件表示當前獲得 KahaDB 讀寫權限的 broker 。
 

優點是性能較好，恢復能力較強，適用于大多數常規場景，對典型的消息使用模式進行了優化 。缺點是在高并發寫入場景下，性能可能會受到一定限制，因為它使用自定義 B - Tree 實現來索引獨寫日志 。
 

2. LevelDB：LevelDB 是從 ActiveMQ 5.8 之后引進的一種基于文件的本地數據庫存儲形式 。配置示例如下：
 

 

<persistenceAdapter>
 

<levelDB directory="activemq-data"/>
 

</persistenceAdapter>
 

它提供比 KahaDB 更快的持久性，因為它不使用自定義 B - Tree 實現來索引獨寫日志，而是使用基于 LevelDB 的索引 。不過，目前 ActiveMQ 官網對于 LevelDB 和 KahaDB 的選擇沒有明確的定論，雖然 LevelDB 性能更優越，但 ActiveMQ 仍然默認使用 KahaDB 。
 

3. JDBC：JDBC 方式是將消息持久化到數據庫中 。配置步驟如下：
 

  • 添加驅動包：將 MySQL 等數據庫的驅動包添加到 ActiveMQ 的lib文件夾 。
 

  • 配置數據源：在activemq.xml文件中添加數據源配置，例如：
 

 

<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">
 

<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
 

<property name="url" value="jdbc:mysql://192.168.1.100:3306/activemq?relaxAutoCommit=true"/>
 

<property name="username" value="root"/>
 

<property name="password" value="root"/>
 

<property name="maxTotal" value="200"/>
 

<property name="poolPreparedStatements" value="true"/>
 

</bean>
 

  • 修改持久化適配器：修改persistenceAdapter元素，使用jdbcPersistenceAdapter，如下：
 

 

<persistenceAdapter>
 

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

</persistenceAdapter>
 

createTablesOnStartup屬性表示是否在啟動時創建數據表，默認值是true 。
 

JDBC 方式的優點是數據存儲在數據庫中，便于管理和查詢，與數據庫相關的工具和技術可以直接應用 。缺點是數據庫的讀寫性能可能不如基于文件的存儲方式，尤其是在高并發場景下，頻繁的數據庫操作可能會成為性能瓶頸，并且依賴于數據庫的穩定性，如果數據庫出現故障，會影響 ActiveMQ 的正常運行 。
 

4. 適用場景分析：如果應用場景對性能要求較高，且數據量不是特別大，KahaDB 和 LevelDB 是比較好的選擇，它們基于文件存儲，讀寫速度相對較快 。對于需要與數據庫緊密結合，或者對數據的管理和查詢有較高要求的場景，JDBC 方式更為合適，例如在一些需要對消息數據進行復雜統計分析的業務中 。
 

（三）客戶端連接與故障轉移
 

  1. failover 協議：客戶端使用failover協議連接到 ActiveMQ 集群中的 broker，該協議提供了故障轉移和自動重連的功能 。連接示例如下：
 

 

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://broker1:61616,tcp://broker2:61616,tcp://broker3:61616)?randomize=false&priorityBackup=true");
 

在這個示例中，failover協議后面的括號內列出了多個 broker 的地址和端口，客戶端會嘗試依次連接這些 broker 。
 

2. 故障轉移機制原理：當客戶端使用failover協議連接到集群時，它會按照配置的順序嘗試連接各個 broker。如果當前連接的 broker 出現故障，客戶端會立即檢測到連接中斷，然后根據配置的策略，自動嘗試連接到其他可用的 broker 。例如，在上述示例中，如果broker1出現故障，客戶端會自動嘗試連接broker2，如果broker2也不可用，會繼續嘗試連接broker3 。
 

randomize=false表示按列表順序調用，不隨機選擇節點；priorityBackup=true表示當randomize為false時，優先選擇第一個節點作為主節點 。這樣，在集群中某個節點出現故障時，客戶端能夠快速切換到其他可用節點，保證消息的發送和接收不受影響，從而提高了系統的可用性和穩定性 。
 

3. 配置參數說明：除了上述的randomize和priorityBackup參數外，還有一些其他常用的配置參數 。initialReconnectDelay參數用于設置首次重連的延遲時間，單位是毫秒，例如initialReconnectDelay=1000表示首次重連延遲 1 秒，這可以避免在網絡瞬間波動時頻繁重連 。maxReconnectAttempts參數用于設置最大重連次數，默認值為-1，表示無限次重連，如果設置為一個具體的數值，如maxReconnectAttempts=5，則當重連次數達到 5 次后，客戶端將不再嘗試重連 。useExponentialBackOff參數用于設置是否使用指數退避算法來調整重連間隔，設置為true時，重連間隔會隨著重連次數的增加而指數級增長，避免在故障未恢復時過于頻繁地重連，影響系統性能 。
 

（四）性能優化建議
 

  1. 內存配置：合理配置 ActiveMQ 的內存參數對于提升性能至關重要。可以通過修改ACTIVEMQ_OPTS環境變量來調整 JVM 的堆內存大小。例如，在啟動 ActiveMQ 的腳本中，將ACTIVEMQ_OPTS設置為-Xms512m -Xmx1024m，表示初始堆內存為 512MB，最大堆內存為 1024MB 。如果系統中消息量較大，并且需要處理大量的并發請求，可以適當增加堆內存的大小，但也要注意不要設置過大，以免導致垃圾回收時間過長，影響系統的響應速度 。同時，可以根據實際情況調整新生代和老年代的比例，例如使用-XX:NewRatio=2表示新生代和老年代的比例為 1:2，以優化垃圾回收的性能 。
 

  1. 線程池調整：ActiveMQ 使用線程池來處理消息的接收、發送和其他任務。可以通過修改activemq.xml配置文件中的線程池參數來優化性能 。例如，調整transportConnectors中的executor參數，增加線程池的大小，以提高處理并發連接的能力 。默認情況下，executor的配置如下：
 

 

<transportConnectors>
 

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600">
 

<transportParam name="executor" value="threadPool"/>
 

</transportConnector>
 

</transportConnectors>
 

可以在broker元素中添加threadPool的配置，如：
 

 

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

<bean id="threadPool" class="org.apache.activemq.util.ServiceStoppingTaskExecutor">
 

<property name="corePoolSize" value="20"/>
 

<property name="maximumPoolSize" value="100"/>
 

<property name="keepAliveTime" value="30000"/>
 

</bean>
 

...
 

</broker>
 

其中，corePoolSize表示線程池的核心線程數，maximumPoolSize表示線程池的最大線程數，keepAliveTime表示線程在空閑狀態下的存活時間，單位是毫秒 。根據系統的負載情況，合理調整這些參數，可以提高 ActiveMQ 對并發請求的處理能力 。
 

3. 網絡優化：在網絡方面，確保服務器之間的網絡帶寬充足，減少網絡延遲。可以通過優化網絡拓撲結構，使用高速網絡設備，如千兆網卡、高性能交換機等，來提高網絡傳輸速度 。同時，合理配置防火墻規則，開放 ActiveMQ 所需的端口，避免端口被阻塞導致通信失敗 。在 ActiveMQ 的配置中，可以調整transportConnectors的一些網絡相關參數，如wireFormat.maxInactivityDuration參數，用于設置連接在多長時間內沒有數據傳輸后會被關閉，適當增大這個值可以避免因短暫的網絡波動而導致連接關閉 。例如：
 

 

<transportConnectors>
 

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600&wireFormat.maxInactivityDuration=30000"/>
 

</transportConnectors>
 

這里將wireFormat.maxInactivityDuration設置為 30000 毫秒（30 秒） 。另外，還可以啟用 TCP 的一些優化參數，如TCP_NODELAY，減少網絡延遲，在transportConnector的uri中添加wireFormat.tcpNoDelay=true即可 。
 

六、集群測試與驗證
 

（一）測試工具與環境搭建
 

  1. 測試工具選擇：在測試 ActiveMQ 集群時，JMeter 是一款常用且功能強大的測試工具。它基于 Java 開發，能夠對各種服務進行性能測試，包括 ActiveMQ 消息中間件。JMeter 提供了豐富的組件和插件，方便用戶進行各種場景的測試。例如，它可以模擬大量的生產者和消費者并發訪問 ActiveMQ 集群，測試集群在高并發情況下的性能表現 。
 

  1. 環境搭建步驟：
 

  •  
    • 安裝 Java 環境：由于 JMeter 是基于 Java 的工具，首先需要確保系統中安裝了 Java 環境。如果尚未安裝，按照前面介紹的方法，安裝 JDK 1.8 及以上版本，并配置好 JAVA_HOME、PATH 和 CLASSPATH 等環境變量 。
     
 

  •  
    • 下載與安裝 JMeter：從 JMeter 官方網站（Apache JMeter - Download Apache JMeter ）下載最新版本的 JMeter 安裝包。下載完成后，解壓安裝包到指定目錄，例如/usr/local/jmeter 。
     
 

  •  
    • 配置 ActiveMQ 相關依賴：為了讓 JMeter 能夠與 ActiveMQ 進行通信，需要將 ActiveMQ 的客戶端庫添加到 JMeter 的類路徑中。將 ActiveMQ 安裝目錄下的lib/activemq-all-x.x.x.jar（x.x.x為 ActiveMQ 的版本號）文件復制到 JMeter 安裝目錄下的lib目錄中 。
     
 

  •  
    • 啟動 JMeter：進入 JMeter 的bin目錄，執行jmeter.sh（Linux 系統）或jmeter.bat（Windows 系統）命令啟動 JMeter。啟動成功后，會出現 JMeter 的圖形化界面，在該界面中可以進行各種測試場景的配置 。
     
 

（二）功能測試
 

  1. 消息的發送與接收測試：在 JMeter 中創建一個線程組，設置線程數為 10（可根據實際情況調整），表示模擬 10 個并發的生產者或消費者。在線程組中添加一個 JMS Point to Point Sampler，配置如下：
 

  •  
    • QueueConnectionFactory：填寫 ActiveMQ 的連接工廠地址，例如tcp://192.168.1.100:61616（根據實際的 ActiveMQ 服務器地址和端口修改） 。
     
 

  •  
    • JNDI name Request queue：指定發送消息的隊列名稱，如testQueue 。
     
 

  •  
    • Content：設置發送的消息內容，例如 “Hello, ActiveMQ Cluster!” 。
     
 

  •  
    • Initial Context Factory：填寫org.apache.activemq.jndi.ActiveMQInitialContextFactory 。
     
 

  •  
    • Provider URL：同樣填寫 ActiveMQ 的服務器地址和端口 。
     
 

啟動測試后，觀察 JMeter 的結果樹，查看是否成功發送和接收消息。如果在結果樹中能夠看到發送的消息內容以及接收的響應，說明消息的發送與接收功能正常 。
 

  1. 持久化消息處理測試：修改上述測試配置，將消息的發送模式設置為持久化模式。在 JMS Point to Point Sampler 中，找到Delivery Mode選項，設置為PERSISTENT 。然后發送一定數量的持久化消息，例如 100 條。發送完成后，關閉 ActiveMQ 集群中的一個節點，再重新啟動該節點。啟動成功后，再次使用 JMeter 進行消息接收測試，檢查是否能夠接收到之前發送的持久化消息。如果能夠正常接收，說明持久化消息處理功能正常，即使節點故障重啟，持久化的消息也不會丟失 。
 

  1. 集群節點間消息同步測試：在集群中創建多個隊列，例如queue1、queue2、queue3 。使用 JMeter 分別向不同節點上的隊列發送消息，然后在其他節點上嘗試接收這些消息。例如，向節點 1 上的queue1發送消息，然后在節點 2 和節點 3 上接收queue1的消息。如果能夠在其他節點上成功接收到消息，說明集群節點間的消息同步功能正常，消息能夠在集群中的各個節點之間正確傳遞 。
 

（三）高可用測試
 

  1. 模擬 broker 節點故障：在 ActiveMQ 集群運行過程中，使用命令行工具或系統管理工具，停止集群中的一個 broker 節點。例如，在 Linux 系統中，可以使用kill -9命令停止 ActiveMQ 進程對應的 PID 。
 

  1. 驗證故障轉移功能：在停止一個 broker 節點后，立即使用 JMeter 進行消息發送和接收測試。觀察 JMeter 的測試結果，查看是否能夠正常發送和接收消息。同時，查看 ActiveMQ 集群的日志文件，確認是否有其他節點成功接管了故障節點的工作。如果 JMeter 能夠正常發送和接收消息，且日志中顯示其他節點已接管故障節點的任務，說明集群的自動故障轉移功能正常，在節點故障時能夠保證消息服務的連續性 。
 

  1. 驗證消息一致性：在故障轉移完成后，檢查集群中各個節點上的消息隊列狀態和消息內容。確保在故障轉移過程中，沒有消息丟失或重復，消息的順序和內容與故障前保持一致。可以通過查詢 ActiveMQ 的管理界面、數據庫（如果使用 JDBC 持久化）或其他相關工具來驗證消息的一致性 。
 

（四）性能測試
 

  1. 測試指標設定：在 JMeter 中配置性能測試場景，設定測試指標，如吞吐量（TPS，Transactions Per Second）、響應時間（Response Time）、并發性能（Concurrent Performance）等 。設置線程組的線程數從 10 逐漸增加到 100，模擬不同并發程度的訪問。設置測試的持續時間為 10 分鐘，以獲取足夠的測試數據 。
 

  1. 測試結果分析：啟動性能測試后，JMeter 會生成詳細的測試報告。在測試報告中，分析吞吐量指標，如果隨著并發數的增加，吞吐量逐漸上升并趨于穩定，說明集群能夠有效地處理并發請求；如果吞吐量在某個并發數下出現明顯下降，可能表示集群已經達到性能瓶頸 。分析響應時間指標，如果平均響應時間隨著并發數的增加而逐漸增加，但仍在可接受范圍內，說明集群的響應性能良好；如果響應時間突然大幅增加，可能存在性能問題，需要進一步排查，如網絡延遲、線程池耗盡等 。分析并發性能指標，觀察集群在高并發情況下的穩定性，是否出現連接超時、消息丟失等異常情況。通過對這些測試結果的分析，可以評估 ActiveMQ 集群的性能表現，并根據分析結果對集群進行優化和調整 。
 

七、常見問題與解決方案
 

在搭建和使用 ActiveMQ 集群的過程中，可能會遇到各種問題，以下是一些常見問題及對應的解決方案：
 

（一）連接失敗
 

  1. 問題描述：客戶端無法連接到 ActiveMQ 集群，報出連接超時或拒絕連接的錯誤。
 

  1. 原因分析：
 

  •  
    • ActiveMQ 服務未啟動或端口被占用。例如，在啟動 ActiveMQ 時，由于系統中其他進程占用了 ActiveMQ 默認的 61616 端口，導致 ActiveMQ 無法正常監聽該端口，從而使客戶端無法連接 。
     
 

  •  
    • 防火墻或安全組攔截了客戶端與 ActiveMQ 之間的通信端口。在一些服務器環境中，防火墻默認會阻止外部對某些端口的訪問，如果沒有開放 ActiveMQ 所需的端口，就會導致連接失敗 。
     
 

  •  
    • 配置的 IP 地址或端口與實際服務不匹配。比如在客戶端的連接配置中，錯誤地填寫了 ActiveMQ 服務器的 IP 地址或端口號，就無法建立有效的連接 。
     
 

  1. 解決方案：
 

  •  
    • 檢查 ActiveMQ 狀態，在 Linux 系統中，可以使用ps -ef | grep activemq命令查看 ActiveMQ 進程是否存在；在 Windows 系統中，可以通過任務管理器查看 ActiveMQ 服務是否運行 。如果服務未啟動，根據前面介紹的啟動方法啟動 ActiveMQ。
     
 

  •  
    • 驗證端口監聽，在 Linux 系統中，使用netstat -tlnp | grep 端口號命令查看指定端口是否處于監聽狀態；在 Windows 系統中，可以使用 TCPView 等工具查看端口占用情況 。如果端口被占用，找到占用端口的進程并停止它，或者修改 ActiveMQ 的配置，使用其他未被占用的端口 。
     
 

  •  
    • 調整防火墻規則，臨時關閉防火墻測試（生產環境慎用），如果關閉防火墻后能夠正常連接，說明是防火墻的問題。在生產環境中，應該開放 ActiveMQ 所需的端口，如在 CentOS 系統中，使用firewall - cmd --zone = public --add - port = 61616/tcp --permanent命令開放 TCP 協議的 61616 端口，使用firewall - cmd --zone = public --add - port = 8161/tcp --permanent命令開放 Web Console 的 8161 端口，然后執行firewall - cmd --reload命令使配置生效 。
     
 

  •  
    • 核對連接配置，確保客戶端代碼中的連接 URL 與 ActiveMQ 服務器的實際配置一致，包括傳輸協議、IP 地址和端口號等 。例如，如果 ActiveMQ 配置了 SSL 連接，客戶端也需要相應地配置 SSL 相關參數 。
     
 

（二）消息丟失
 

  1. 問題描述：在消息的發送和接收過程中，出現消息丟失的情況，導致業務數據不一致。
 

  1. 原因分析：
 

  •  
    • 網絡問題，網絡不穩定或者網絡斷開可能導致消息在傳輸過程中丟失。比如在網絡波動較大的環境中，消息在從生產者發送到 ActiveMQ 服務器，或者從服務器發送到消費者的過程中，可能會因為網絡中斷而丟失 。
     
 

  •  
    • 系統崩潰，ActiveMQ 服務宕機或者客戶端崩潰也可能導致消息丟失。如果在消息尚未持久化到存儲介質時，ActiveMQ 服務器突然崩潰，內存中的消息就會丟失；同樣，客戶端在接收消息后但尚未處理完成時崩潰，也可能導致消息丟失 。
     
 

  •  
    • 配置問題，錯誤的配置（如不恰當的持久化設置）可能導致消息在未被處理前丟失。例如，將消息設置為非持久化模式，且在消息處理過程中出現異常，就可能導致消息丟失 。
     
 

  •  
    • 其他原因，消息隊列滿、錯誤的消息處理邏輯等，也可能導致消息丟失。如果消息隊列達到了最大容量，新的消息將無法進入隊列，從而導致丟失；另外，如果消費者在處理消息時出現錯誤，沒有進行適當的異常處理，也可能導致消息被錯誤地丟棄 。
     
 

  1. 解決方案：
 

  •  
    • 消息持久化，使用 KahaDB、LevelDB 或 JDBC 等持久化方式，確保消息在服務器崩潰等情況下不會丟失 。例如，配置 KahaDB 持久化：
     
 

 

<persistenceAdapter>
 

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

</persistenceAdapter>
 

  • 消息確認機制，使用合適的消息確認模式，如 CLIENT_ACKNOWLEDGE，消費者接收到消息后需要顯式調用消息的 acknowledge 方法來確認，避免在消息處理過程中丟失 。在代碼中，設置消息確認模式的示例如下：
 

 

Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
 

  • 事務管理，在消息處理過程中使用事務，確保消息被完整處理。例如，在發送多條消息時，將這些消息放在一個事務中，只有當所有消息都成功發送后，事務才會提交，否則回滾，保證消息的一致性 。
 

 

Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
 

try {
 

// 發送多條消息的代碼
 

session.commit();
 

} catch (Exception e) {
 

session.rollback();
 

}
 

  • 網絡連接和異常處理，優化網絡連接，使用重試機制和合理的異常處理策略可以減少因網絡問題導致的消息丟失 。在客戶端代碼中，可以使用failover協議來實現自動重連和故障轉移，并且在捕獲到網絡異常時，進行適當的重試操作 。例如：
 

 

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://broker1:61616,tcp://broker2:61616,tcp://broker3:61616)?randomize=false&priorityBackup=true");
 

  • 配置優化，合理配置消息隊列的大小、消費者的數量等參數，避免系統過載 。在activemq.xml文件中，可以設置destinationPolicy來調整隊列的相關參數，如concurrentConsumers設置并發消費者的數量，prefetch設置預取限制等 。
 

 

<destinationPolicy>
 

<policyMap>
 

<policyEntries>
 

<policyEntry queue=">" concurrentConsumers="10" prefetch="100"/>
 

</policyEntries>
 

</policyMap>
 

</destinationPolicy>
 

（三）性能瓶頸
 

  1. 問題描述：ActiveMQ 集群在高并發情況下，出現吞吐量下降、響應時間延長等性能問題。
 

  1. 原因分析：
 

  •  
    • 網絡延遲，在分布式場景下，網絡延遲對 ActiveMQ 性能的影響尤為明顯，特別是在跨地理位置部署時更為顯著。例如，客戶端與 ActiveMQ 服務器之間的網絡帶寬不足，導致消息傳輸速度緩慢 。
     
 

  •  
    • 磁盤 I/O 瓶頸，Broker 在處理持久化消息時，需要頻繁地進行磁盤讀寫操作，如果磁盤性能不佳，將極大限制消息的處理效率。比如使用傳統的機械硬盤，其讀寫速度遠低于固態硬盤，在高并發寫入消息時，容易出現 I/O 瓶頸 。
     
 

  •  
    • 內存管理限制，不合理的 JVM 設置可能導致頻繁的垃圾回收，從而影響到消息的處理速度。如果 JVM 堆內存設置過小，在處理大量消息時，容易導致內存溢出，引發頻繁的垃圾回收，使系統響應變慢 。
     
 

  •  
    • 消息持久化策略不佳，選擇不合適的持久化策略，或者持久化配置不合理，會影響消息的發送和消費效率。例如，使用 JDBC 持久化時，如果數據庫配置不當，可能會導致數據庫連接池耗盡，從而影響消息的持久化速度 。
     
 

  1. 解決方案：
 

  •  
    • 網絡優化，使用高質量的網絡設備和線路，確保網絡通暢，減少網絡延遲。例如，升級網絡帶寬，使用千兆或萬兆網絡；優化網絡拓撲結構，減少網絡跳數 。同時，可以啟用 TCP 的一些優化參數，如TCP_NODELAY，減少網絡延遲，在transportConnector的uri中添加wireFormat.tcpNoDelay=true即可 。
     
 

 

<transportConnectors>
 

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600&wireFormat.tcpNoDelay=true"/>
 

</transportConnectors>
 

  • 磁盤性能提升，采用 RAID 技術提升磁盤性能和數據安全，或者使用高性能的 SSD 硬盤，以減少數據寫入的延遲 。同時，合理規劃磁盤的 I/O 容量，避免瓶頸 。在 ActiveMQ 的配置中，可以調整transportConnectors的一些網絡相關參數，如wireFormat.maxInactivityDuration參數，用于設置連接在多長時間內沒有數據傳輸后會被關閉，適當增大這個值可以避免因短暫的網絡波動而導致連接關閉 。
 

 

<transportConnectors>
 

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600&wireFormat.maxInactivityDuration=30000"/>
 

</transportConnectors>
 

  • 內存優化，根據業務量調整 JVM 堆內存大小，使用現代 GC 算法，如 G1，減少 GC 停頓時間 。在啟動 ActiveMQ 的腳本中，可以通過修改ACTIVEMQ_OPTS環境變量來調整 JVM 的堆內存大小 。例如：
 

 

export ACTIVEMQ_OPTS="-Xms512m -Xmx1024m -XX:+UseG1GC"
 

  • 持久化優化，選擇合適的持久化策略，并進行相應的配置優化。如果使用 JDBC 持久化，選擇支持高并發寫入操作的數據庫，如 MySQL、Oracle 等，并調整數據庫參數，如連接池大小、緩存策略等，以提高寫入性能 。如果使用 KahaDB 或 LevelDB 持久化，可以啟用異步寫盤，通過配置使得消息的存儲過程異步執行，降低磁盤 I/O 對性能的影響 。
 

 

<persistenceAdapter>
 

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

</persistenceAdapter>
 

（四）消息堆積
 

  1. 問題描述：消息的生產速度遠大于消費速度，導致消息在隊列中大量堆積，最終耗盡存儲資源。
 

  1. 原因分析：
 

  •  
    • 消費者處理能力不足，消費者的業務邏輯復雜，或者消費者的數量過少，導致無法及時處理接收到的消息 。例如，在一個訂單處理系統中，消費者在處理訂單消息時，需要進行復雜的業務校驗和數據庫操作，導致處理一條消息的時間較長，從而造成消息堆積 。
     
 

  •  
    • 消息生產突發高峰，在某些業務場景下，如電商促銷活動、限時搶購等，會出現消息生產的突發高峰，而系統沒有足夠的處理能力來應對 。
     
 

  •  
    • 消費邏輯錯誤，消費者的消費邏輯中存在死循環、異常處理不當等問題，導致消費者無法正常消費消息 。比如消費者在處理消息時，捕獲到異常后沒有進行適當的處理，而是直接退出循環，導致后續消息無法被消費 。
     
 

  1. 解決方案：
 

  •  
    • 增加消費者數量，通過增加并發消費者的數量，可以提高消息的并行消費能力 。在activemq.xml文件中，可以通過destinationPolicy配置來設置并發消費者的數量 。
     
 

 

<destinationPolicy>
 

<policyMap>
 

<policyEntries>
 

<policyEntry queue=">" concurrentConsumers="10"/>
 

</policyEntries>
 

</policyMap>
 

</destinationPolicy>
 

  • 優化消費邏輯，簡化消費者的業務邏輯，減少不必要的操作，提高消息處理速度 。同時，完善消費邏輯中的異常處理，確保在出現異常時，消費者能夠繼續正常消費消息 。
 

 

try {
 

// 消費消息的業務邏輯
 

} catch (Exception e) {
 

// 異常處理邏輯，記錄日志，進行重試等操作
 

}
 

  • 流量削峰，采用消息隊列的流量削峰機制，如設置消息的過期時間，避免無限制堆積 。在activemq.xml文件中，可以為隊列設置timeToLive屬性，指定消息的過期時間 。
 

 

<destinationPolicy>
 

<policyMap>
 

<policyEntries>
 

<policyEntry queue=">" timeToLive="60000"/> <!-- 消息60秒后過期 -->
 

</policyEntries>
 

</policyMap>
 

</destinationPolicy>
 

  • 消息批量處理，消費者可以采用批量處理的方式，一次從隊列中獲取多個消息進行處理，減少與 ActiveMQ 服務器的交互次數，提高處理效率 。在客戶端代碼中，可以設置prefetch參數來調整一次預取的消息數量 。
 

 

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://broker:61616?jms.prefetchPolicy.all=100");
 

八、總結與展望
 

通過本文的介紹，我們全面深入地了解了 ActiveMQ 集群搭建與高可用方案設計的相關知識。從基礎概念入手，詳細闡述了 ActiveMQ 的核心概念，包括消息、隊列、主題、生產者和消費者等，這些概念是理解和使用 ActiveMQ 的基石 。接著，對集群搭建前的準備工作進行了詳細說明，涵蓋了環境準備、下載與安裝 ActiveMQ 以及配置文件的解讀，為后續的集群搭建奠定了堅實的基礎 。
 

在集群搭建實戰部分，深入探討了 Master - Slave 模式和 Broker - Cluster 模式的搭建方法，并針對生產環境推薦了 Master - Slave + Broker - Cluster 的組合配置，這種配置方式充分發揮了兩種模式的優勢，既保證了高可用性，又實現了負載均衡 。在高可用方案設計與優化方面，詳細介紹了 Zookeeper 在 ActiveMQ 集群中的應用，包括選舉 Master 節點和維護集群一致性；分析了消息持久化策略，如 KahaDB、LevelDB 和 JDBC 等的特點和適用場景；講解了客戶端連接與故障轉移的機制，以及相關的配置參數；還提出了一系列性能優化建議，如內存配置、線程池調整和網絡優化等 。
 

在集群測試與驗證環節，介紹了使用 JMeter 進行功能測試、高可用測試和性能測試的方法，通過測試可以及時發現集群中存在的問題，并進行針對性的優化 。最后，對搭建和使用 ActiveMQ 集群過程中可能遇到的常見問題，如連接失敗、消息丟失、性能瓶頸和消息堆積等，進行了原因分析，并給出了相應的解決方案 。
 

展望未來，隨著分布式系統和微服務架構的不斷發展，消息中間件的需求將持續增長。ActiveMQ 作為一款成熟的開源消息中間件，也將不斷演進和發展。未來，ActiveMQ 有望在性能優化、功能擴展和與新興技術的融合等方面取得更大的突破 。在性能優化方面，將進一步提升消息的處理速度和吞吐量，降低延遲，以滿足高并發、低延遲的業務需求 。在功能擴展方面，可能會增加對更多消息協議的支持，提供更豐富的消息處理功能，如消息過濾、消息轉換等 。在與新興技術的融合方面，ActiveMQ 可能會更好地與云計算、容器技術、大數據等技術相結合，為企業提供更全面、高效的消息通信解決方案 。同時，隨著安全和隱私問題的日益重要，ActiveMQ 也將加強安全機制的建設，確保消息的安全傳輸和存儲 。總之，ActiveMQ 在未來的消息中間件領域仍將發揮重要作用，為分布式系統的發展提供有力支持 。











本文來自互聯網用戶投稿，該文觀點僅代表作者本人，不代表本站立場。本站僅提供信息存儲空間服務，不擁有所有權，不承擔相關法律責任。

如若轉載，請注明出處：http://www.pswp.cn/bicheng/79514.shtml


繁體地址，請注明出處：http://hk.pswp.cn/bicheng/79514.shtml

英文地址，請注明出處：http://en.pswp.cn/bicheng/79514.shtml


如若內容造成侵權/違法違規/事實不符，請聯系多彩編程網進行投訴反饋email:809451989@qq.com，一經查實，立即刪除！











相關文章










【項目歸檔】數據抓取+GenAI+數據分析








【項目歸檔】數據抓取+GenAI+數據分析




年后這兩個月頻繁組織架構變動&#xff0c;所以博客很久沒更新。現在暫時算是塵埃落定&#xff0c;趁這段時間整理一下。 入職九個月&#xff0c;自己參與的項目有4個&#xff0c;負責前后端開發&#xff0c;測試&#xff0c;devops&#xff08;全棧/doge&#xff09;&#xff…




閱讀更多...

















服務器熱備份，服務器熱備份的方法有哪些？








服務器熱備份，服務器熱備份的方法有哪些？




服務器熱備份是保障業務連續性的重要技術手段&#xff0c;通過實時數據同步和快速故障切換&#xff0c;確保在主服務器故障時備份服務器能無縫接管。以下是常見的服務器熱備份方法及其核心要點&#xff1a; 雙機熱備&#xff08;Active-Standby/Active-Active&#xff09; 
主備…




閱讀更多...

















【AI平臺】n8n入門6：調用MCP服務（非社區節點）








【AI平臺】n8n入門6：調用MCP服務（非社區節點）




前言 
前邊用n8n搭建一個MCP服務&#xff0c;現在&#xff0c;用n8n調用其他服務商提供的MCP服務。本文以高德地圖服務為例&#xff0c;記錄一下操作過程。 
實現案例功能 
MCP是啥 
MCP&#xff08;Model Context Protocol&#xff0c;模型上下文協議&#xff09;是由Anthropi…




閱讀更多...

















基于ArduinoIDE的任意型號單片機 + GPS北斗BDS衛星定位








基于ArduinoIDE的任意型號單片機 + GPS北斗BDS衛星定位




提示&#xff1a;文章寫完后&#xff0c;目錄可以自動生成&#xff0c;如何生成可參考右邊的幫助文檔 文章目錄 前言1.1 器件選擇1.2 接線方案 二、驅動實現2.1 核心代碼解析&#xff08;arduino/ESP32-S3&#xff09; 三、坐標解析代碼四、典型問題排查總結 前言 
北斗衛星導航…




閱讀更多...

















經典算法 最小生成樹(prim算法)








經典算法 最小生成樹(prim算法)




最小生成樹 
題目描述 
給定一個 n 個點 m 條邊的無向圖&#xff0c;圖中可能存在重邊和自環&#xff0c;邊權可能為負數。 
求最小生成樹的樹邊權重之和。如果最小生成樹不存在&#xff0c;則輸出 impossible。 
給定一張邊帶權的無向圖 G  (V, E)&#xff0c;其中&#xff1a…




閱讀更多...

















LeetCode算法題 （設計鏈表）Day16！！！C/C++








LeetCode算法題 （設計鏈表）Day16！！！C/C++




https://leetcode.cn/problems/design-linked-list/description/ 
一、題目分析 你可以選擇使用單鏈表或者雙鏈表&#xff0c;設計并實現自己的鏈表。 
單鏈表中的節點應該具備兩個屬性&#xff1a;val 和 next 。val 是當前節點的值&#xff0c;next 是指向下一個節點的指針/引…




閱讀更多...

















《解鎖GCC版本升級：開啟編程新世界大門》








《解鎖GCC版本升級：開啟編程新世界大門》




《解鎖GCC版本升級：開啟編程新世界大門》 一、引言：GCC 版本升級的魔法鑰匙 
在編程的廣闊天地里，GCC（GNU Compiler Collection）宛如一座燈塔，為無數開發者照亮前行的道路。它是一款開源且功能強大的編譯器集合，支持 C、C++、Objective - C、Fortran、Ada 等多種編程語言…




閱讀更多...

















toLua筆記








toLua筆記




基本 
LuaState luaStatenew LuaState();
luaState.Start();
luaState.DoString("xxx");
luaState.DoFile("yyy.lua");
luaState.Require("zzz");//不要加.lua后綴
luaState.CheckTop();//檢查解析器棧頂為空
luaState.Dispose();
luaStatenull;…




閱讀更多...

















go實現雙向鏈表








go實現雙向鏈表




需求 
實現雙向鏈表的節點生成、正反向遍歷、指定刪除。 
實現 
package mainimport ("fmt"
)type zodiac_sign struct {number   intdizhi    stringanimal   stringyear     intprevious *zodiac_signnext     *zodiac_sign
}// 添加
// func add_node_by_order(pr…




閱讀更多...

















AI實踐指南：AGENT、RAG和MCP在Java中的簡單實現








AI實踐指南：AGENT、RAG和MCP在Java中的簡單實現




在當今AI快速發展的時代&#xff0c;有幾個核心概念正在改變我們構建智能應用的方式。本文將用簡單易懂的語言介紹三個重要概念&#xff1a;AGENT&#xff08;AI代理&#xff09;、RAG&#xff08;檢索增強生成&#xff09;和MCP&#xff08;多通道感知&#xff09;&#xff0c…




閱讀更多...

















解決VMware虛擬機能搜索到網頁但打不開的問題








解決VMware虛擬機能搜索到網頁但打不開的問題




&#x1f334; 問題描述 
很奇怪&#xff0c;不知道為什么&#xff0c;我安裝的Windows 10虛擬機能在瀏覽器中搜索到網頁&#xff0c;但點擊具體的網頁鏈接就是死活不能加載出來&#xff0c;如下圖所示&#xff1a; 點擊第一個鏈接&#xff0c;加載了四五分鐘&#xff0c;結果就…




閱讀更多...

















JVM性能調優的基礎知識 | JVM內部優化與運行時優化








JVM性能調優的基礎知識 | JVM內部優化與運行時優化




目錄 
JVM內部的優化邏輯 
JVM的執行引擎 
解釋執行器 
即時編譯器 
JVM采用哪種方式&#xff1f; 
即時編譯器類型 
JVM的分層編譯5大級別&#xff1a; 
分層編譯級別&#xff1a; 
熱點代碼&#xff1a; 
如何找到熱點代碼&#xff1f; 
java兩大計數器&#xff1a; 
OSR 編譯…




閱讀更多...

















什么是多租戶系統








什么是多租戶系統




隨著云計算和 SaaS&#xff08;Software as a Service&#xff09;模式的普及&#xff0c;多租戶架構&#xff08;Multi-Tenant Architecture&#xff09;成為 SaaS 產品設計中的核心模式之一。多租戶架構允許多個用戶&#xff08;租戶&#xff09;共享同一套基礎設施和應用&am…




閱讀更多...

















多線程系列三：這就是線程的狀態？








多線程系列三：這就是線程的狀態？




1.認識線程的狀態 
NEW&#xff1a;Thread對象已經創建好了&#xff0c;但還沒有調用start方法在系統中創建線程 RUNNABLE&#xff1a;就緒狀態&#xff0c;表示這個線程正在CPU上執行&#xff0c;或準備就緒&#xff0c;隨時可以去CPU上執行 BLOCKED&#xff1a;表示由于鎖競爭…




閱讀更多...

















【C語言練習】019. 使用結構體數組存儲復雜數據








【C語言練習】019. 使用結構體數組存儲復雜數據




019. 使用結構體數組存儲復雜數據 019. 使用結構體數組存儲復雜數據示例1&#xff1a;定義一個結構體并創建結構體數組定義結構體創建并初始化結構體數組輸出結果 示例2&#xff1a;動態輸入數據到結構體數組定義結構體動態輸入數據示例輸入和輸出 示例3&#xff1a;使用結構體…




閱讀更多...

















**Java面試大冒險：謝飛機的幽默與技術碰撞記**








**Java面試大冒險：謝飛機的幽默與技術碰撞記**




互聯網大廠Java求職者面試&#xff1a;一場嚴肅與搞笑交織的技術盛宴 
場景&#xff1a; 互聯網大廠面試間 
人物&#xff1a; 
面試官&#xff1a; 一位嚴肅的資深架構師&#xff0c;對技術要求嚴格。謝飛機&#xff1a; 一位搞笑的程序員&#xff0c;技術實力參差不齊。 第一…




閱讀更多...

















MySQL進階(三)








MySQL進階(三)




五、鎖 
1. 概述 
鎖是計算機協調多個進程或線程并發訪問某一資源的機制&#xff08;避免爭搶&#xff09;。 
在數據庫中&#xff0c;除傳統的計算資源&#xff08;如 CPU、RAM、I/O 等&#xff09;的爭用以外&#xff0c;數據也是一種供許多用戶共享的資源。如何保證數據并發…




閱讀更多...

















【BLE】【nRF Connect】 精講nRF Connect自動化測試套件（宏錄制、XML腳本）








【BLE】【nRF Connect】 精講nRF Connect自動化測試套件（宏錄制、XML腳本）




目錄 
前言 
1. nRF Connect自動化測試介紹 
1.1. nRF connect宏錄制功能介紹 
1.2. 電腦端XML方式 
1.3  實際應用案例 
1.3.1 BLE 穩定性測試 
1.3.2 設備固件更新（DFU）測試 
1.3.3 批量設備配置 
1.4 操作步驟 
1.5 注意事項 
2. nRF Connect日志記錄 
2.1. 日志記錄功能 …




閱讀更多...

















【數據結構】堆的完整實現








【數據結構】堆的完整實現




堆的完整實現 堆的完整實現GitHub地址前言堆的核心功能實現重溫堆的定義堆結構定義1. 堆初始化與銷毀2. 元素交換函數3. 堆化操作向上調整&#xff08;子→父&#xff09;向下調整&#xff08;父→子&#xff09; 4. 堆元素插入5. 堆元素刪除6. 輔助功能函數堆的判空獲取堆頂元…




閱讀更多...

















如何優化MySQL主從復制的性能？








如何優化MySQL主從復制的性能？




優化MySQL主從復制的性能需要從硬件、配置、架構設計和運維策略等多方面入手。以下是詳細的優化方案&#xff1a; 
一、減少主庫寫入壓力 
1. ?主庫優化? 
二進制日志&#xff08;binlog&#xff09;優化?&#xff1a; 
使用 binlog_formatROW 以獲得更高效的復制和更少的數…




閱讀更多...


















最新文章


















Copyright @ 2022~2023