ActiveMQ 介紹
ActiveMQ 是一款高性能、開源的消息中間件，支持多種消息協議（如 JMS、AMQP、MQTT 等），能夠實現應用程序之間的異步通信和消息傳遞。它提供點對點（Queue）和發布/訂閱（Topic）兩種消息模型，支持消息持久化、事務處理、消息優先級、延遲消息等功能，確保消息的可靠傳輸和順序處理。此外，ActiveMQ 還具備集群、主從復制等高可用性特性，以及靈活的配置和擴展能力，適用于構建分布式系統和微服務架構，幫助開發者構建高效、可靠的消息驅動應用。
ActiveMQ 監控是確保消息中間件高可用性和性能的關鍵環節。通過監控關鍵指標，如連接數、消費者和生產者數量、隊列和主題的入隊/出隊速率、內存和存儲使用率、以及死信隊列消息數量等，可以及時發現并解決潛在問題。
觀測云
觀測云是一款專為 IT 工程師打造的全鏈路可觀測產品，它集成了基礎設施監控、應用程序性能監控和日志管理，為整個技術棧提供實時可觀察性。這款產品能夠幫助工程師全面了解端到端的用戶體驗追蹤，了解應用內函數的每一次調用，以及全面監控云時代的基礎設施。此外，觀測云還具備快速發現系統安全風險的能力，為數字化時代提供安全保障。
部署 DataKit
DataKit 是一個開源的、跨平臺的數據收集和監控工具，由觀測云開發并維護。它旨在幫助用戶收集、處理和分析各種數據源，如日志、指標和事件，以便進行有效的監控和故障排查。DataKit 支持多種數據輸入和輸出格式，可以輕松集成到現有的監控系統中。
登錄觀測云控制臺，在「集成」 - 「DataKit」選擇對應安裝方式，當前采用 Linux 主機部署 DataKit。
數據采集
ActiveMQ 配置
ActiveMQ 使用 Java 語言編寫，支持通過 jmx prometheus 插件進行指標暴露。
- 下載 jmx-exporter
下載地址：GitHub - prometheus/jmx_exporter: A process for collecting metrics using JMX MBeans for Prometheus consumption
- 新增 ActiveMQ jmx 配置
在 ActiveMQ 的 conf 目錄下新增?jmx.yaml?文件，內容如下：
lowercaseOutputName: true
lowercaseOutputLabelNames: true
blacklistObjectNames:- "org.apache.activemq:clientId=*,*"
whitelistObjectNames:- "org.apache.activemq:destinationType=Queue,*"- "org.apache.activemq:destinationType=Topic,*"- "org.apache.activemq:type=Broker,brokerName=*"- "org.apache.activemq:type=Topic,brokerName=*"- "org.apache.activemq:type=Broker,brokerName=*,destinationType=Queue,destinationName=*,endpoint=*,clientId=*,consumerId=*"- "org.apache.activemq:type=Broker,brokerName=*,destinationType=Topic,destinationName=*,endpoint=*,clientId=*,consumerId=*"
rules:
- pattern: org.apache.activemq:type=Broker,brokerName=(\S+),destinationType=(\S+),destinationName=(\S+),endpoint=(\S+),clientId=(\S+),consumerId=(\S+),?>(\w+)name: "activemq_consumer_$7"labels:broker_name: "$1"destination_type: "$2"destination_name: "$3"endpoint: "$4"client_id: "$5"consumer_id: "$6"- pattern: org.apache.activemq<type=Broker, brokerName=(\S*), destinationType=Queue, destinationName=(\S*)><>(\w+)name: activemq_queue_$3attrNameSnakeCase: truelabels:broker_name: "$1"destination_name: "$2"destination_type: "Queue"- pattern: org.apache.activemq<type=Broker, brokerName=(\S*), destinationType=Topic, destinationName=(\S*)><>(\w+)name: activemq_topic_$3attrNameSnakeCase: truelabels:broker_name: "$1"destination_name: "$2"destination_type: "Topic"- pattern: org.apache.activemq<type=Broker, brokerName=(\S*)><>CurrentConnectionsCountname: activemq_connectionstype: GAUGElabels:broker_name: "$1"connection_type: current
- pattern: org.apache.activemq<type=Broker, brokerName=(\S*)><>TotalConnectionsCountname: activemq_connectionstype: GAUGElabels:broker_name: "$1"connection_type: total- pattern: org.apache.activemq<type=Broker, brokerName=(\S*)><>Total(.*)Countname: activemq_$2_totaltype: COUNTERlabels:broker_name: "$1"- pattern: org.apache.activemq<type=Broker, brokerName=(\S*)><>(.*)PercentUsagename: activemq_$2_usage_ratiotype: GAUGElabels:broker_name: "$1"valueFactor: 0.01
- 調整 ActiveMQ 啟動參數
在 ActiveMQ 的 bin 目錄下，修改 setenv 文件，添加如下內容：
ACTIVEMQ_OPTS="-javaagent:${ACTIVEMQ_BASE}/jmx_prometheus_javaagent-1.2.0.jar=8081:${ACTIVEMQ_BASE}/conf/jmx.yaml"
當前設置的端口號為 8081，下面采集會用到，可以進行調整，但需要跟采集端口保持一致。
- 重啟 ActiveMQ
bin/activemq stop
bin/activemq start
DataKit 開啟 Prom 采集器
開啟?prometheus?采集器采集 ActiveMQ 的指標。進入 DataKit 安裝目錄?/usr/local/datakit/conf.d，復制 prom 目錄下的配置文件并命名為?activemq.conf。 示例如下：
cp prom.conf.sample activemq.conf
調整內容如下：
[[inputs.prom]]## Exporter URLs.urls = ["http://localhost:8081/metrics"]source = "activemq"
...
重啟 DataKit
調整完畢后，重啟 DataKit。
datakit service -R
關鍵指標
| 指標名稱 | 描述 | 單位 |
|---|---|---|
connections | 當前與 ActiveMQ Broker 建立的客戶端連接數量 | 個 |
consumer_total | 累計創建的消費者（Consumer）數量 | 個 |
dequeue_total | 所有隊列/主題中成功被消費的消息總數 | 條 |
enqueue_total | 所有隊列/主題中接收到的新消息總數 | 條 |
jobschedulerstore_usage_ratio | JobSchedulerStore 的使用比例，表示持久化任務存儲空間占用 | 百分比（%） |
memory_usage_ratio | ActiveMQ Broker 使用的內存占配置上限的比例 | 百分比（%） |
message_total | 當前在 Broker 上存在的所有消息數量（包括待處理和未確認的消息） | 條 |
producer_total | 累計創建的消息生產者（Producer）數量 | 個 |
queue_always_retroactive | 表示該隊列是否啟用“始終回溯”功能，允許新消費者獲取歷史消息 | 布爾值（0/1） |
queue_average_blocked_time | 隊列因資源不足而阻塞的時間平均值 | 毫秒（ms） |
queue_average_enqueue_time | 消息進入隊列所需的平均時間 | 毫秒（ms） |
queue_average_message_size | 隊列中消息的平均字節大小 | 字節（B） |
queue_blocked_producer_warning_interval | 生產者因隊列滿而被阻塞的警告間隔時間 | 秒（s） |
queue_blocked_sends | 因隊列滿而導致生產者被阻塞的累計次數 | 次 |
queue_cache_enabled | 表示當前隊列是否啟用了緩存機制 | 布爾值（0/1） |
queue_consumer_count | 當前訂閱或監聽該隊列的消費者數量 | 個 |
queue_cursor_full | 表示隊列的消息游標（Cursor）是否已達到最大容量 | 布爾值（0/1） |
queue_cursor_memory_usage | 隊列游標使用的內存量 | 字節（B） |
queue_cursor_percent_usage | 隊列游標內存使用占總分配內存的比例 | 百分比（%） |
queue_dequeue_count | 從該隊列中成功消費的消息數量 | 條 |
queue_dispatch_count | 已經嘗試派發給消費者的累計消息數量 | 條 |
queue_dlq | 被轉移到死信隊列（DLQ）中的消息數量 | 條 |
queue_duplicate_from_store_count | 由于網絡或其他問題導致消息從持久化存儲中重復讀取的次數 | 條 |
queue_enqueue_count | 添加到該隊列的消息數量 | 條 |
queue_expired_count | 因設置 TTL 而過期的消息數量 | 條 |
queue_forward_count | 該隊列將消息轉發到其他目的地的累計次數 | 條 |
queue_in_flight_count | 正在被消費者處理但尚未確認的消息數量 | 條 |
queue_max_audit_depth | 審計重復消息時的最大記錄條目數 | 條 |
queue_max_enqueue_time | 消息進入隊列所需時間的最大值 | 毫秒（ms） |
queue_max_message_size | 隊列中單條消息的最大字節大小 | 字節（B） |
queue_max_page_size | 分頁加載消息時的最大頁大小 | 條 |
queue_max_producers_to_audit | 可以被追蹤并用于重復檢查的生產者最大數量 | 個 |
queue_memory_limit | 該隊列為消息存儲所分配的最大內存 | 字節（B） |
queue_memory_percent_usage | 隊列當前內存使用占配置上限的比例 | 百分比（%） |
queue_memory_usage_byte_count | 隊列當前實際占用的內存大小 | 字節（B） |
queue_memory_usage_portion | 隊列在整體內存使用中的占比 | 浮點數（0~1） |
queue_min_enqueue_time | 消息進入隊列所需時間的最小值 | 毫秒（ms） |
queue_min_message_size | 隊列中單條消息的最小字節大小 | 字節（B） |
queue_paused | 表示該隊列是否處于暫停狀態 | 布爾值（0/1） |
queue_prioritized_messages | 表示是否啟用消息優先級排序功能 | 布爾值（0/1） |
queue_producer_count | 當前向該隊列發送消息的生產者數量 | 個 |
queue_producer_flow_control | 是否啟用生產者流量控制 | 布爾值（0/1） |
queue_queue_size | 當前隊列中等待被消費的消息總數 | 條 |
queue_send_duplicate_from_store_to_dlq | 從持久化存儲中發送到死信隊列的重復消息數量 | 條 |
queue_store_message_size | 隊列中所有消息在持久化存儲中的總大小 | 字節（B） |
queue_temp_usage_limit | 隊列臨時消息存儲的最大限制 | 字節（B） |
queue_temp_usage_percent_usage | 隊列臨時存儲使用占配置上限的比例 | 百分比（%） |
queue_total_blocked_time | 隊列因資源不足而導致的累計阻塞時間 | 毫秒（ms） |
queue_use_cache | 表示該隊列是否啟用了緩存機制 | 布爾值（0/1） |
store_usage_ratio | 主消息存儲（如文件系統或數據庫）的使用比例 | 百分比（%） |
temp_usage_ratio | 臨時消息存儲的使用比例 | 百分比（%） |
topic_always_retroactive | 表示該主題是否啟用“始終回溯”功能，允許新消費者獲取歷史消息 | 布爾值（0/1） |
topic_average_blocked_time | 主題因資源不足而阻塞的時間平均值 | 毫秒（ms） |
topic_average_enqueue_time | 消息進入主題所需的平均時間 | 毫秒（ms） |
topic_average_message_size | 主題中消息的平均字節大小 | 字節（B） |
topic_blocked_producer_warning_interval | 生產者因主題滿而被阻塞的警告間隔時間 | 秒（s） |
topic_blocked_sends | 因主題滿而導致生產者被阻塞的累計次數 | 次 |
topic_cache_enabled | 表示當前主題是否啟用了緩存機制 | 布爾值（0/1） |
topic_consumer_count | 當前訂閱或監聽該主題的消費者數量 | 個 |
topic_cursor_full | 表示主題的消息游標（Cursor）是否已達到最大容量 | 布爾值（0/1） |
topic_cursor_memory_usage | 主題游標使用的內存量 | 字節（B） |
topic_cursor_percent_usage | 主題游標內存使用占總分配內存的比例 | 百分比（%） |
topic_dequeue_count | 從該主題中成功消費的消息數量 | 條 |
topic_dispatch_count | 已經嘗試派發給消費者的累計消息數量 | 條 |
topic_dlq | 被轉移到死信主題（DLQ）中的消息數量 | 條 |
topic_duplicate_from_store_count | 由于網絡或其他問題導致消息從持久化存儲中重復讀取的次數 | 條 |
topic_enqueue_count | 添加到該主題的消息數量 | 條 |
topic_expired_count | 因設置 TTL 而過期的消息數量 | 條 |
topic_forward_count | 該主題將消息轉發到其他目的地的累計次數 | 條 |
topic_in_flight_count | 正在被消費者處理但尚未確認的消息數量 | 條 |
topic_max_audit_depth | 審計重復消息時的最大記錄條目數 | 條 |
topic_max_enqueue_time | 消息進入主題所需時間的最大值 | 毫秒（ms） |
topic_max_message_size | 主題中單條消息的最大字節大小 | 字節（B） |
topic_max_page_size | 分頁加載消息時的最大頁大小 | 條 |
topic_max_producers_to_audit | 可以被追蹤并用于重復檢查的生產者最大數量 | 個 |
topic_memory_limit | 該主題為消息存儲所分配的最大內存 | 字節（B） |
topic_memory_percent_usage | 主題當前內存使用占配置上限的比例 | 百分比（%） |
topic_memory_usage_byte_count | 主題當前實際占用的內存大小 | 字節（B） |
topic_memory_usage_portion | 主題在整體內存使用中的占比 | 浮點數（0~1） |
topic_min_enqueue_time | 消息進入主題所需時間的最小值 | 毫秒（ms） |
topic_min_message_size | 主題中單條消息的最小字節大小 | 字節（B） |
topic_prioritized_messages | 表示是否啟用消息優先級排序功能 | 布爾值（0/1） |
topic_producer_count | 當前向該主題發送消息的生產者數量 | 個 |
topic_producer_flow_control | 是否啟用生產者流量控制 | 布爾值（0/1） |
topic_queue_size | 當前主題中等待被消費的消息總數 | 條 |
topic_send_duplicate_from_store_to_dlq | 從持久化存儲中發送到死信主題的重復消息數量 | 條 |
topic_store_message_size | 主題中所有消息在持久化存儲中的總大小 | 字節（B） |
topic_temp_usage_limit | 主題臨時消息存儲的最大限制 | 字節（B） |
topic_temp_usage_percent_usage | 主題臨時存儲使用占配置上限的比例 | 百分比（%） |
topic_total_blocked_time | 主題因資源不足而導致的累計阻塞時間 | 毫秒（ms） |
topic_use_cache | 表示該主題是否啟用了緩存機制 | 布爾值（0/1） |
場景視圖
登錄觀測云控制臺，點擊「場景」 -「新建儀表板」，輸入 “ActiveMQ”， 選擇 “ActiveMQ”，點擊 “確定” 即可添加視圖。
通過概覽，可以了解到 ActiveMQ 的基礎信息，如消息總積壓、連接數等。
Queue 可以看到隊列消息的生產、消費、積壓等。
Topic 部分可以了解到 Topic 的消息的生產、消費、積壓等。
監控器（告警）
觀測云內置了監控器模板，可以選擇從模版創建監控器，并開啟適合業務的監控器以及時通知相關成員關注問題，觸發條件、頻率等信息可以依據實際業務進行調整。
登錄觀測云控制臺，點擊「監控」 -「新建監控器」，輸入 “ActiveMQ”， 選擇對應的監控器，點擊 “確定” 即可添加。
ActiveMQ 隊列中有大量被阻塞的發送操作
ActiveMQ 隊列消息積壓
觀測云內置了10多個 ActiveMQ 告警監控器，可按實際需求添加。
總結
使用觀測云采集 ActiveMQ 指標數據，可以實現自動化監控、數據收集和告警通知。這不僅有助于優化資源分配，還能提高系統的可靠性和響應速度，確保消息的高效傳遞和處理。
：RAG下的ETL快速上手)
