如何用Python及PyMQI实现IBM MQ实时消息监听器？

完全可以基于PyMQI实现实时获取消息的监听器，下面分享两种常用的实践方案及注意事项：

方案1：轮询式监听（简单易上手） #

这种方式通过循环调用get()方法并设置超时，实现近似实时的消息获取，适合对实时性要求不是极高的场景，代码简单易维护。

import pymqi

# MQ连接配置
queue_manager = "QMGR_NAME"
channel = "CHANNEL_NAME"
host = "MQ_HOST"
port = "MQ_PORT"
queue_name = "TARGET_QUEUE"

# 建立MQ连接
conn_info = f"{host}({port})"
qmgr = pymqi.connect(queue_manager, channel, conn_info)
queue = pymqi.Queue(qmgr, queue_name)

try:
    print("开始监听队列...")
    while True:
        # 设置1秒超时，超时后继续循环等待
        message = queue.get(None, pymqi.MQMI_NONE, 1000)
        if message:
            print(f"收到消息: {message.decode('utf-8')}")
            # 在这里添加你的消息处理逻辑
finally:
    # 确保资源释放
    queue.close()
    qmgr.disconnect()

• 关键注意：设置合理的超时时间，避免无限制阻塞；循环中要捕获pymqi.MQError异常，防止单次获取失败导致监听终止。

方案2：异步回调式监听（高效实时） #

利用PyMQI的MQCB回调机制，注册消息到达时的处理函数，由MQ主动通知，这种方式实时性更强、资源利用率更高。

import pymqi
import pymqi.mqconstants as mqc

def message_handler(msg):
    """消息到达时的回调处理函数"""
    print(f"异步收到消息: {msg.decode('utf-8')}")
    # 在这里添加你的消息处理逻辑
    return mqc.MQCC_OK

# MQ连接配置
queue_manager = "QMGR_NAME"
channel = "CHANNEL_NAME"
host = "MQ_HOST"
port = "MQ_PORT"
queue_name = "TARGET_QUEUE"

conn_info = f"{host}({port})"
qmgr = pymqi.connect(queue_manager, channel, conn_info)
queue = pymqi.Queue(qmgr, queue_name)

try:
    print("启动异步监听...")
    # 注册回调函数，设置监听模式为异步获取
    queue.callback(message_handler, mqc.MQOP_GET)
    # 保持主线程运行，等待回调触发
    while True:
        pymqi.wait(1000)
finally:
    queue.close()
    qmgr.disconnect()

• 关键注意：主线程需持续运行，否则回调机制会终止；回调函数内要处理异常，避免影响后续消息监听；部分MQ版本需提前开启异步支持配置。

通用实践注意事项 #

• 异常捕获：必须捕获pymqi.MQError及子类异常，处理队列断开、权限不足、消息格式错误等场景，保证监听稳定性。
• 消息确认：根据业务需求选择合适的消息确认模式（如MQGMO_SYNCPOINT），处理完消息后正确提交或回滚，避免消息丢失或重复消费。
• 资源管理：监听终止时务必关闭队列、断开连接，防止MQ资源泄漏。

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