从IBM WebSphere MQ队列取消息发REST API及本地WMQ问题排查

核心问题诊断 #

你的代码存在几个关键问题，导致消息无法被正确读取和移除：

• 空通道名称：客户端连接WMQ必须指定有效的服务器连接通道，空值会导致连接逻辑异常
• 不必要的认证参数：若队列管理器未启用认证，传递用户名密码会干扰正常连接流程
• 事务会话处理不严谨：异常场景下未回滚事务，且消息内容解析方式无法展示实际数据
• 无接收超时：receive()无限等待可能导致脚本看似执行完成但实际未获取到消息

分步修复方案 #

1. 配置有效服务器连接通道 #

默认队列管理器会创建SYSTEM.DEF.SVRCONN通道，将代码中的通道名称改为：

def channelName = "SYSTEM.DEF.SVRCONN"

2. 移除不必要的认证参数 #

无需认证时使用无参的createConnection()方法：

def conn = cf.createConnection()

3. 优化事务会话逻辑 #

保留事务会话时，需在异常场景下添加回滚操作；若无需事务，可改为自动确认模式：

// 非事务模式（可选）
def sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE)

4. 正确解析消息内容 #

message.toString()仅返回元数据，需根据消息类型获取实际内容：

if (message instanceof javax.jms.TextMessage) {
    def textContent = ((javax.jms.TextMessage) message).getText()
    logger.info("Received message content: ${textContent}")
}

5. 添加接收超时避免挂起 #

为receive()设置超时时间，避免脚本无限等待：

def message = consumer.receive(5000) // 5秒超时

修复后的完整代码 #

@Grapes([
@Grab(group='com.ibm.mq', module='com.ibm.mq.allclient', version='9.1.0.0'),
@Grab(group='org.slf4j', module='slf4j-api', version='1.7.32'),
@Grab(group='org.slf4j', module='slf4j-simple', version='1.7.32')
 ])

 import com.ibm.msg.client.jms.JmsConnectionFactory
 import com.ibm.msg.client.jms.JmsFactoryFactory
 import com.ibm.msg.client.wmq.WMQConstants
 import javax.jms.Connection
 import javax.jms.Message
 import javax.jms.MessageConsumer
 import javax.jms.Queue
 import javax.jms.Session
 import javax.jms.TextMessage
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory

 def logger = LoggerFactory.getLogger(getClass())

 // 连接配置
 def hostName = "127.0.0.1"
 def hostPort = 1414
 def channelName = "SYSTEM.DEF.SVRCONN"
 def queueManagerName = "QM1"
 def queueName = "connector"

 Connection conn = null
 Session sess = null
 MessageConsumer consumer = null

 try {
    def ff = JmsFactoryFactory.getInstance(WMQConstants.WMQ_PROVIDER)
    def cf = ff.createConnectionFactory()

    cf.setStringProperty(WMQConstants.WMQ_HOST_NAME, hostName)
    cf.setIntProperty(WMQConstants.WMQ_PORT, hostPort)
    cf.setStringProperty(WMQConstants.WMQ_CHANNEL, channelName)
    cf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT)
    cf.setStringProperty(WMQConstants.WMQ_QUEUE_MANAGER, queueManagerName)

    conn = cf.createConnection()
    sess = conn.createSession(true, Session.SESSION_TRANSACTED)
    def destination = sess.createQueue(queueName)

    consumer = sess.createConsumer(destination)
    logger.info("CONSUMER: ${consumer}")

    conn.start()

    // 5秒内读取消息
    def message = consumer.receive(5000)
    if (message != null) {
        // 解析文本消息
        if (message instanceof TextMessage) {
            def textContent = ((TextMessage) message).getText()
            logger.info("Received text message: ${textContent}")
            
            // 此处添加发送到REST API的逻辑
            // 示例：使用Groovy HTTP客户端发送请求
            // def restClient = new groovyx.net.http.RESTClient("https://your-api-endpoint/")
            // def response = restClient.post(body: textContent, requestContentType: "application/json")
            // logger.info("REST API response status: ${response.status}")
        } else {
            logger.info("Received non-text message metadata: ${message.toString()}")
        }
        // 提交事务移除消息
        sess.commit()
        logger.info("Transaction committed, message removed from queue")
    } else {
        logger.info("No message found in 5 seconds")
    }
 } catch (Exception e) {
    // 异常时回滚事务
    if (sess != null) {
        try {
            sess.rollback()
            logger.info("Transaction rolled back due to error")
        } catch (Exception rollbackEx) {
            rollbackEx.printStackTrace()
        }
    }
    e.printStackTrace()
 } finally {
    // 确保资源关闭
    if (consumer != null) consumer.close()
    if (sess != null) sess.close()
    if (conn != null) conn.close()
    logger.info("JMS resources closed")
 }

额外验证步骤 #

• 用WMQ Explorer确认SYSTEM.DEF.SVRCONN通道处于运行状态
• 检查队列权限：确保连接用户对connector队列拥有读取和删除权限
• 查看队列管理器错误日志：排查连接或权限相关的异常信息

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