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快速上手)
