Python3.7如何读取ActiveMQ.Advisory消息属性获取consumerCount并解决STOMP订阅异常

1 解决STOMP订阅异常新建队列的问题 #

ActiveMQ的STOMP协议实现默认会将无前缀的目的地识别为队列，订阅主题必须在目的地路径前添加/topic/前缀。你原来的订阅路径没有前缀，所以被判定为队列创建请求，修改TOPIC变量即可：

# 原错误写法
# TOPIC = 'ActiveMQ.Advisory.Consumer.Topic.xbox_001'
# 修正后写法
TOPIC = '/topic/ActiveMQ.Advisory.Consumer.Topic.xbox_001'

2 读取consumerCount属性的方法 #

ActiveMQ会将Advisory消息的所有自定义JMS属性直接映射为STOMP消息的Header字段，不需要读取消息体（Advisory消息体本身为空属于正常情况），直接从回调的message.headers中取值即可，修正后的完整可运行代码如下：

import time
import stomp

class MyListener(stomp.ConnectionListener):
    def on_error(self, headers, message):
        print('received an error "%s"' % message)
    def on_message(self, message):
        # 直接从headers读取consumerCount
        if 'consumerCount' in message.headers:
            consumer_count = int(message.headers['consumerCount'])
            print(f"当前活跃消费者数量：{consumer_count}")
            # 此处添加你的业务逻辑：消费者数量为0时调用snap7触发PLC告警

# 注意路径前缀
TOPIC = '/topic/ActiveMQ.Advisory.Consumer.Topic.xbox_001'

conn = stomp.Connection()
conn.set_listener('', MyListener())
conn.connect('admin', 'password', wait=True)
conn.subscribe(destination=TOPIC, id='1', ack='auto')

time.sleep(3600)
conn.disconnect()

内容的提问来源于stack exchange，提问作者Simon Dreyer