如何使用 Python 获取 ActiveMQ 的队列列表信息？

需要对 ActiveMQ 监控，获取节点上的所有队列名，生产者，消费者个数以及消息的个数，用 python 做。
麻烦问下哪个库有这个功能？或者有其他什么方法能处理？
看了 stomp 的 api，只能收发消息。上网搜了一圈，也没合适的答案。都是 Java 的。
麻烦各位，多谢


如何使用 Python 获取 ActiveMQ 的队列列表信息？
3 回复 
import stomp

class ActiveMQListener(stomp.ConnectionListener):
    def on_error(self, frame):
        print(f'Received an error: {frame.body}')
    
    def on_message(self, frame):
        print(f'Received message: {frame.body}')

def get_queue_list(host='localhost', port=61613, username='admin', password='admin'):
    """
    获取ActiveMQ队列列表信息
    需要安装stomp.py: pip install stomp.py
    """
    try:
        # 创建STOMP连接
        conn = stomp.Connection([(host, port)])
        conn.set_listener('', ActiveMQListener())
        
        # 连接到ActiveMQ
        conn.connect(username, password, wait=True)
        
        # 发送管理请求获取队列信息
        # ActiveMQ管理消息的目标地址
        management_queue = 'ActiveMQ.Advisory.Queue'
        
        # 订阅管理队列
        conn.subscribe(destination=management_queue, id=1, ack='auto')
        
        # 发送获取队列列表的请求
        conn.send(
            body='',
            destination='ActiveMQ.Advisory.Queue',
            headers={'reply-to': management_queue}
        )
        
        # 等待一段时间接收响应
        import time
        time.sleep(2)
        
        # 断开连接
        conn.disconnect()
        
        print("注意：此方法需要ActiveMQ启用Advisory Messages")
        print("在activemq.xml中添加：<advisorySupport>true</advisorySupport>")
        
    except Exception as e:
        print(f"连接失败: {e}")

# 使用方法
if __name__ == '__main__':
    # 默认连接本地ActiveMQ
    get_queue_list()
    
    # 连接远程ActiveMQ
    # get_queue_list(host='your-activemq-server', username='your-user', password='your-pass')

替代方案：使用REST API（如果启用了Web Console）：

import requests
from requests.auth import HTTPBasicAuth

def get_queues_via_rest(host='localhost', port=8161, username='admin', password='admin'):
    """
    通过ActiveMQ REST API获取队列信息
    """
    url = f'http://{host}:{port}/api/jolokia/read/org.apache.activemq:type=Broker,brokerName=localhost'
    
    try:
        response = requests.get(
            url,
            auth=HTTPBasicAuth(username, password),
            headers={'Content-Type': 'application/json'}
        )
        
        if response.status_code == 200:
            data = response.json()
            # 解析队列信息
            queues = data.get('value', {}).get('Queues', [])
            for queue in queues:
                print(f"队列: {queue.get('objectName')}")
        else:
            print(f"请求失败: {response.status_code}")
            
    except Exception as e:
        print(f"错误: {e}")

# 使用REST API
# get_queues_via_rest()

总结：推荐使用REST API方式，更稳定可靠。


ActiveMQ 有个 Web 控制台，上面可以展示各种队列、统计，配置信息。按一下 F12，看一下是向服务器发的什么请求，应该能找到线索。。

多谢，写了个爬虫直接从 WEB 控制台爬取相关数据了。

回到顶部