IBM MQ队列破坏性读取（清理）后仍残留消息的问题

问题根源分析 #

你的代码存在几个关键问题，导致偶发的消息残留：

1. 未显式启用破坏性读取
   虽然用MQOO_INPUT_SHARED打开队列，但未在MQGetMessageOptions中指定CMQC.MQGMO_DESTRUCTIVE，MQ默认的读取行为在某些场景下可能不会直接删除消息（比如消息处于锁定状态时），导致消息残留。

2. 异常处理逻辑覆盖不全
   仅处理了MQRC_TRUNCATED_MSG_ACCEPTED警告，未针对MQRC_NO_MSG_AVAILABLE（2033）这个标准的无消息返回码做判断，当遇到其他非致命异常时，会直接跳出循环，遗漏未读取的消息。

3. 缺少无等待与同步点设置
   未添加CMQC.MQGMO_NO_WAIT会导致get操作在队列暂时空的时候阻塞，可能触发非预期的异常退出；如果队列是事务性的，缺少同步点设置会导致读取操作未提交，消息在程序退出后回滚残留。

4. 队列深度读取的时序问题
   getCurrentDepth()在get后打印，但该值是队列实时深度，若在get和读取深度之间有其他进程写入（或事务回滚的消息），会导致日志显示深度为0但实际有残留。

修复后的代码 #

void purge() {
    MQQueueManager qMgr = null;
    MQQueue queue = null;
    try {
        qMgr = new MQQueueManager(queueManager);
        int openOptions = CMQC.MQOO_INPUT_SHARED + CMQC.MQOO_INQUIRE + CMQC.MQOO_FAIL_IF_QUIESCING;
        MQGetMessageOptions gmo = new MQGetMessageOptions();
        // 添加破坏性读取、无等待、同步点、禁止静默等关键选项
        gmo.options = CMQC.MQGMO_FAIL_IF_QUIESCING 
                    + CMQC.MQGMO_ACCEPT_TRUNCATED_MSG 
                    + CMQC.MQGMO_DESTRUCTIVE 
                    + CMQC.MQGMO_NO_WAIT 
                    + CMQC.MQGMO_SYNCPOINT;

        queue = qMgr.accessQueue(queueName, openOptions);
        
        int msgCount = 0;
        while (true) {
            try {
                MQMessage receiveMsg = new MQMessage();
                queue.get(receiveMsg, gmo);
                msgCount++;
                qMgr.commit(); // 提交事务，确保消息被永久删除
            } catch (MQException e) {
                if ((e.completionCode == CMQC.MQCC_WARNING) && (e.reasonCode == CMQC.MQRC_TRUNCATED_MSG_ACCEPTED)) {
                    LOG.info("已检测并删除截断消息");
                    msgCount++;
                    qMgr.commit();
                } else if (e.reasonCode == CMQC.MQRC_NO_MSG_AVAILABLE) {
                    LOG.info("队列已无剩余消息");
                    break;
                } else {
                    LOG.error("清理队列时发生异常: " + e.getMessage(), e);
                    break;
                }
            }
            LOG.info("当前队列剩余消息数: " + queue.getCurrentDepth());
        }
        LOG.info("共清理 " + msgCount + " 条消息");
    } catch (MQException e) {
        LOG.error("连接队列管理器或打开队列失败: " + e.getMessage(), e);
        // 异常时回滚未提交事务
        if (qMgr != null) {
            try {
                qMgr.backout();
            } catch (MQException be) {
                LOG.error("回滚事务失败: " + be.getMessage(), be);
            }
        }
    } finally {
        // 确保资源被安全释放
        if (queue != null) {
            try {
                queue.close();
            } catch (MQException e) {
                LOG.error("关闭队列失败: " + e.getMessage(), e);
            }
        }
        if (qMgr != null) {
            try {
                qMgr.disconnect();
            } catch (MQException e) {
                LOG.error("断开队列管理器连接失败: " + e.getMessage(), e);
            }
        }
    }
}

关键修改说明 #

• 添加MQGMO_DESTRUCTIVE：强制每次get操作都是破坏性删除，彻底移除消息。
• 添加MQGMO_NO_WAIT：当队列无消息时立即返回，避免阻塞导致的异常退出。
• 事务同步与提交/回滚：针对事务性队列，确保读取操作被持久化，避免因未提交导致的消息回滚。
• 完善异常处理：明确捕获MQRC_NO_MSG_AVAILABLE作为循环终止条件，其他异常打印日志后退出，避免遗漏消息。
• 资源安全释放：使用finally块确保队列和管理器资源被正确关闭，避免资源泄漏。

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