ActiveMQ 可观测性最佳实践
ActiveMQ 介绍
ActiveMQ 是一款高性能、开源的消息中间件，支持多种消息协议（如 JMS、AMQP、MQTT 等），能够实现应用程序之间的异步通信和消息传递。它提供点对点（Queue）和发布/订阅（Topic）两种消息模型，支持消息持久化、事务处理、消息优先级、延迟消息等功能，确保消息的可靠传输和顺序处理。此外，ActiveMQ 还具备集群、主从复制等高可用性特性，以及灵活的配置和扩展能力，适用于构建分布式系统和微服务架构，帮助开发者构建高效、可靠的消息驱动应用。
ActiveMQ 监控是确保消息中间件高可用性和性能的关键环节。通过监控关键指标，如连接数、消费者和生产者数量、队列和主题的入队/出队速率、内存和存储使用率、以及死信队列消息数量等，可以及时发现并解决潜在问题。
观测云
观测云是一款专为 IT 工程师打造的全链路可观测产品，它集成了基础设施监控、应用程序性能监控和日志管理，为整个技术栈提供实时可观察性。这款产品能够帮助工程师全面了解端到端的用户体验追踪，了解应用内函数的每一次调用，以及全面监控云时代的基础设施。此外，观测云还具备快速发现系统安全风险的能力，为数字化时代提供安全保障。
部署 DataKit
DataKit 是一个开源的、跨平台的数据收集和监控工具，由观测云开发并维护。它旨在帮助用户收集、处理和分析各种数据源，如日志、指标和事件，以便进行有效的监控和故障排查。DataKit 支持多种数据输入和输出格式，可以轻松集成到现有的监控系统中。
登录观测云控制台，在「集成」 - 「DataKit」选择对应安装方式，当前采用 Linux 主机部署 DataKit。
数据采集
ActiveMQ 配置
ActiveMQ 使用 Java 语言编写，支持通过 jmx prometheus 插件进行指标暴露。
- 下载 jmx-exporter
下载地址：https://github.com/prometheus/jmx_exporter
- 新增 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_$3
attrNameSnakeCase: true
labels:
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_$3
attrNameSnakeCase: true
labels:
broker_name: "$1"
destination_name: "$2"
destination_type: "Topic"
- pattern: org.apache.activemq<type=Broker, brokerName=(\S*)><>CurrentConnectionsCount
name: activemq_connections
type: GAUGE
labels:
broker_name: "$1"
connection_type: current
- pattern: org.apache.activemq<type=Broker, brokerName=(\S*)><>TotalConnectionsCount
name: activemq_connections
type: GAUGE
labels:
broker_name: "$1"
connection_type: total
- pattern: org.apache.activemq<type=Broker, brokerName=(\S*)><>Total(.*)Count
name: activemq_$2_total
type: COUNTER
labels:
broker_name: "$1"
- pattern: org.apache.activemq<type=Broker, brokerName=(\S*)><>(.*)PercentUsage
name: activemq_$2_usage_ratio
type: GAUGE
labels:
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 指标数据，可以实现自动化监控、数据收集和告警通知。这不仅有助于优化资源分配，还能提高系统的可靠性和响应速度，确保消息的高效传递和处理。
