Java中如何检查IBMQ消息是否已推送，避免队列重复？

完全可以实现队列消息的去重，基于你提供的transactionReference唯一标识，下面给出几种可行的落地方案，结合你的现有代码进行改造：

方案一：直接扫描队列检查（适合小队列场景） #

通过JMS的QueueBrowser遍历队列中的消息，解析后对比transactionReference判断是否已存在。

步骤1：添加消息存在性检查方法 #

private boolean isMessageExistsInQueue(String queueName, String transactionReference) {
    try {
        // 创建队列浏览器（只读遍历，不会消费消息）
        Queue queue = (Queue) jmsTemplate.getConnectionFactory()
                .createConnection().createSession().createQueue(queueName);
        QueueBrowser browser = jmsTemplate.getConnectionFactory()
                .createConnection().createSession().createBrowser(queue);
        Enumeration<?> messageEnum = browser.getEnumeration();
        ObjectMapper objectMapper = new ObjectMapper();

        // 遍历所有消息对比唯一标识
        while (messageEnum.hasMoreElements()) {
            TextMessage message = (TextMessage) messageEnum.nextElement();
            IbmqRequest existingRequest = objectMapper.readValue(message.getText(), IbmqRequest.class);
            if (transactionReference.equals(existingRequest.getTransactionReference())) {
                browser.close();
                return true;
            }
        }
        browser.close();
        return false;
    } catch (Exception e) {
        log.error("检查队列消息存在性失败：队列[{}]", queueName, e);
        return false; // 异常场景可根据业务调整，比如默认允许发送
    }
}

步骤2：改造原发送方法 #

public void sendMessage(String queueName, IbmqRequest ibmqRequest) {
    try {
        String transactionReference = ibmqRequest.getTransactionReference();
        // 先检查消息是否已存在
        if (isMessageExistsInQueue(queueName, transactionReference)) {
            log.info("消息已存在于队列[{}]，transactionReference：{}，跳过发送", queueName, transactionReference);
            return;
        }

        // 正常发布消息
        ObjectMapper objectMapper = new ObjectMapper();
        String jsonMessage = objectMapper.writeValueAsString(ibmqRequest);
        jmsTemplate.convertAndSend(queueName, jsonMessage);
        log.info("消息发送至队列[{}]：{}", queueName, jsonMessage);

    } catch (JsonProcessingException e) {
        log.error("(sendMessage) 转换IbmqRequest为JSON失败", e);
    } catch (Exception e) {
        log.error("(sendMessage) 发送消息至队列[{}]失败", queueName, e);
    }
}

优缺点：无需额外依赖，但队列消息量大时遍历性能差，仅适合消息量少的场景。

方案二：外部缓存记录（适合高并发场景） #

用Redis等缓存工具记录已发送且未被消费的transactionReference，检查时直接查缓存，性能更优。

步骤1：添加缓存操作方法 #

@Autowired
private RedisTemplate<String, String> redisTemplate;

private static final String MESSAGE_CACHE_PREFIX = "ibmq:sent:";

// 检查消息是否已发送
private boolean isMessageAlreadySent(String transactionReference) {
    return redisTemplate.hasKey(MESSAGE_CACHE_PREFIX + transactionReference);
}

// 标记消息为已发送（设置过期时间避免内存泄漏）
private void markMessageAsSent(String transactionReference, long expireMinutes) {
    redisTemplate.opsForValue().set(
            MESSAGE_CACHE_PREFIX + transactionReference,
            "sent",
            expireMinutes,
            TimeUnit.MINUTES
    );
}

// 消息消费完成后移除缓存标记（需在消费逻辑中调用）
public void removeMessageMark(String transactionReference) {
    redisTemplate.delete(MESSAGE_CACHE_PREFIX + transactionReference);
}

步骤2：改造发送方法 #

public void sendMessage(String queueName, IbmqRequest ibmqRequest) {
    try {
        String transactionReference = ibmqRequest.getTransactionReference();
        if (isMessageAlreadySent(transactionReference)) {
            log.info("消息已发送过，transactionReference：{}，跳过发送", transactionReference);
            return;
        }

        ObjectMapper objectMapper = new ObjectMapper();
        String jsonMessage = objectMapper.writeValueAsString(ibmqRequest);
        jmsTemplate.convertAndSend(queueName, jsonMessage);
        log.info("消息发送至队列[{}]：{}", queueName, jsonMessage);

        // 标记为已发送，设置30分钟过期（根据业务调整）
        markMessageAsSent(transactionReference, 30);

    } catch (JsonProcessingException e) {
        log.error("(sendMessage) 转换IbmqRequest为JSON失败", e);
    } catch (Exception e) {
        log.error("(sendMessage) 发送消息至队列[{}]失败", queueName, e);
        // 发送失败时移除缓存标记
        redisTemplate.delete(MESSAGE_CACHE_PREFIX + ibmqRequest.getTransactionReference());
    }
}

优缺点：检查速度快，适合高并发场景，但需要依赖缓存组件，且需保证消费完成后清理缓存。

方案三：利用IBM MQ原生特性（Correlation ID匹配） #

将transactionReference设置为消息的Correlation ID，通过MQ原生API查询是否存在该标识的消息。

步骤1：添加Correlation ID检查方法 #

private boolean isMessageExistsWithCorrelationId(String queueName, String transactionReference) {
    MQQueueManager queueManager = null;
    MQQueue queue = null;
    try {
        // 初始化MQ连接参数（根据实际配置修改）
        MQEnvironment.hostname = "你的MQ主机地址";
        MQEnvironment.port = 1414;
        MQEnvironment.channel = "你的通道名";
        MQEnvironment.userID = "用户名";
        MQEnvironment.password = "密码";

        queueManager = new MQQueueManager("你的队列管理器名");
        int openOptions = MQC.MQOO_INPUT_SHARED | MQC.MQOO_BROWSE;
        queue = queueManager.accessQueue(queueName, openOptions);

        // 设置匹配条件为Correlation ID
        MQMessage message = new MQMessage();
        message.correlationId = transactionReference.getBytes(StandardCharsets.UTF_8);
        
        MQGetMessageOptions gmo = new MQGetMessageOptions();
        gmo.options = MQC.MQGMO_BROWSE_FIRST | MQC.MQGMO_NO_WAIT;
        gmo.matchOptions = MQC.MQMO_MATCH_CORREL_ID;

        try {
            queue.get(message, gmo);
            return true;
        } catch (MQException e) {
            if (e.reasonCode == MQC.MQRC_NO_MSG_AVAILABLE) {
                return false;
            }
            throw e;
        }
    } catch (Exception e) {
        log.error("通过Correlation ID检查消息失败：队列[{}]，标识[{}]", queueName, transactionReference, e);
        return false;
    } finally {
        // 关闭资源
        if (queue != null) try { queue.close(); } catch (MQException ignored) {}
        if (queueManager != null) try { queueManager.disconnect(); } catch (MQException ignored) {}
    }
}

步骤2：改造发送方法 #

public void sendMessage(String queueName, IbmqRequest ibmqRequest) {
    try {
        String transactionReference = ibmqRequest.getTransactionReference();
        if (isMessageExistsWithCorrelationId(queueName, transactionReference)) {
            log.info("消息已存在于队列[{}]，transactionReference：{}，跳过发送", queueName, transactionReference);
            return;
        }

        ObjectMapper objectMapper = new ObjectMapper();
        String jsonMessage = objectMapper.writeValueAsString(ibmqRequest);
        
        // 发送时设置Correlation ID
        jmsTemplate.convertAndSend(queueName, jsonMessage, message -> {
            message.setJMSCorrelationID(transactionReference);
            return message;
        });
        
        log.info("消息发送至队列[{}]：{}", queueName, jsonMessage);

    } catch (JsonProcessingException e) {
        log.error("(sendMessage) 转换IbmqRequest为JSON失败", e);
    } catch (Exception e) {
        log.error("(sendMessage) 发送消息至队列[{}]失败", queueName, e);
    }
}

优缺点：利用MQ原生能力，无需外部依赖，但代码复杂度高，需熟悉IBM MQ原生API。

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