快捷導(dǎo)航

使用Redis防止重復(fù)發(fā)送RabbitMQ消息的方法詳解

更新時(shí)間：2025年01月20日 11:21:02 作者：零臣

今天遇到一個(gè)問(wèn)題,發(fā)送MQ消息的時(shí)候需要保證不會(huì)重復(fù)發(fā)送,注意不是可靠到達(dá),這里保證的是不會(huì)生產(chǎn)多條一樣的消息,所以本文主要介紹了使用Redis防止重復(fù)發(fā)送RabbitMQ消息的方法,需要的朋友可以參考下

問(wèn)題

今天遇到一個(gè)問(wèn)題，發(fā)送MQ消息的時(shí)候需要保證不會(huì)重復(fù)發(fā)送，注意不是可靠到達(dá)（可靠到達(dá)可以通過(guò)消息確認(rèn)機(jī)制和回調(diào)接口保證），這里保證的是不會(huì)生產(chǎn)多條一樣的消息。

方法

綜合討論下來(lái)決定使用Redis緩存來(lái)解決，因?yàn)橄啾扔趯⒂涗洸迦霐?shù)據(jù)庫(kù)Redis更為高效和便捷。

檢驗(yàn)是否已經(jīng)發(fā)送

在發(fā)送消息之前根據(jù)相關(guān)信息組合成key去Redis中查找，找到后檢測(cè)值是否為存在并且是否為設(shè)定的值，若存在且與設(shè)定的值一樣，則返回false，說(shuō)明該消息已經(jīng)發(fā)送過(guò)了。

    public boolean isSend(String messageType, Long bizId, int hashCode) {
        // 根據(jù)消息類型、業(yè)務(wù)id和哈希值組合成key
        String key = this.genKey(messageType, bizId, hashCode);
        Long value = super.get(key);

        if (value != null && value.equals(DEFAULT_VALUE)) {
            return false;
        }

        return true;
    }
   
   /**get方法*/
    public V get(K key) {
        if (key == null) {
            return null;
        } else {
            try {
                // 在key前添加前綴和名字，并將原來(lái)的key進(jìn)行json序列化
                String realKey = this.genRealKey(key);
                String content = (String)this.redisTemplate.opsForValue().get(realKey);
                // 若get到的值不為null則進(jìn)行json反序列化
                return content == null ? null : this.valueSerializer.deserialize(content);
            } catch (Exception e) {
                CACHE.error("", key.toString(), "", "0", e);
                return null;
            }
        }
    }

以上就是檢驗(yàn)消息是否重復(fù)的方法，需要注意的是JSON序列化，因?yàn)镽edis默認(rèn)使用的是JDK序列化，這種序列化后的內(nèi)容不僅多而且不易于閱讀，因此將其改為Json序列化。

發(fā)送后添加緩存

在發(fā)送消息的時(shí)候會(huì)先在Redis中put一個(gè)以相關(guān)信息組合為key，value為默認(rèn)值的記錄，過(guò)期時(shí)間為5min。

    public void sendMessage(String messageType, Long bizId, int hashCode) {
        super.put(genKey(messageType, bizId, hashCode), DEFAULT_VALUE);
    }
    
    /**put方法*/
    public void put(K key, V value) {
        try {
            if (key != null && null != value) {
                // 進(jìn)行json序列化
                String content = this.valueSerializer.serialize(value);
                this.redisTemplate.opsForValue().set(this.genRealKey(key), content, this.expires, this.timeUnit);
            }
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }

發(fā)送消息方法

最后的發(fā)送消息方法大致代碼如下：

    public void sendMQMessage(Long bizId, String messageTemplateCode, String msg, int msgHashCode, String exchange, String routingKey) {
        //加入緩存
        boolean send = true;
        //String messageType = MessageTypeUtil.getMessageType(messageTemplateCode);
        if (bizId != null) {
            // 檢測(cè)是否已經(jīng)發(fā)送
            send = sendMessageCache.isSend(messageTemplateCode, bizId, msgHashCode);
        }

        //發(fā)送mq消息
        if (send) {
            if (bizId != null) {
                // 加入緩存
                sendMessageCache.sendMessage(messageTemplateCode, bizId, msgHashCode);
            }
            // 發(fā)送消息
            messageSender.send(exchange, routingKey, msg);
        }
    }

到此這篇關(guān)于使用Redis防止重復(fù)發(fā)送RabbitMQ消息的方法詳解的文章就介紹到這了,更多相關(guān)Redis防止重復(fù)發(fā)送RabbitMQ內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: