解決rocketmq-client查詢(xún)手動(dòng)發(fā)送消息異常問(wèn)題

更新時(shí)間：2023年08月29日 15:26:27 作者：小淼同學(xué)

這篇文章主要介紹了解決rocketmq-client查詢(xún)手動(dòng)發(fā)送消息異常問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

rocketmq-client查詢(xún)手動(dòng)發(fā)送消息異常

今天處理rocketmq的后臺(tái)的一些問(wèn)題

下面這個(gè)問(wèn)題是你也用網(wǎng)上4.x的rocketmq版本的監(jiān)控后臺(tái)，才會(huì)出現(xiàn)的

MQClientException: CODE: 208 DESC: query message by id finished, but no message.

這個(gè)是我們用rocketmq-client手動(dòng)發(fā)送消息的時(shí)候，再去查看消息詳情的時(shí)候遇到的問(wèn)題，會(huì)報(bào)錯(cuò)，消息找不到

這是由于查詢(xún)出來(lái)的mq消息，這些消息的msgId，不是真的msgId，而是UniqueKey

為什么會(huì)這樣呢，我們來(lái)看一段源碼

public static List<MessageExt> decodes(java.nio.ByteBuffer byteBuffer, final boolean readBody) {
        List<MessageExt> msgExts = new ArrayList<MessageExt>();
        while (byteBuffer.hasRemaining()) {
            MessageExt msgExt = clientDecode(byteBuffer, readBody);
            if (null != msgExt) {
                msgExts.add(msgExt);
            } else {
                break;
            }
        }
        return msgExts;
    }

1.以上是解析從broker獲取到的消息

  public static MessageExt clientDecode(java.nio.ByteBuffer byteBuffer, final boolean readBody) {
        return decode(byteBuffer, readBody, true, true);
    }

2.注意第四個(gè)參數(shù)是true

它代表是否是client端，就是監(jiān)控端

 public static MessageExt decode(
        java.nio.ByteBuffer byteBuffer, final boolean readBody, final boolean deCompressBody, final boolean isClient) {
        try {
            MessageExt msgExt;
            //1.重點(diǎn)看這里
            if (isClient) {
                msgExt = new MessageClientExt();
            } else {
                msgExt = new MessageExt();
            }
            //此處省略
            ...........
            ByteBuffer byteBufferMsgId = ByteBuffer.allocate(MSG_ID_LENGTH);
            String msgId = createMessageId(byteBufferMsgId, msgExt.getStoreHostBytes(), msgExt.getCommitLogOffset());
            msgExt.setMsgId(msgId);
            //2.重點(diǎn)看這里
            if (isClient) {
                ((MessageClientExt) msgExt).setOffsetMsgId(msgId);
            }
            return msgExt;
        } catch (Exception e) {
            byteBuffer.position(byteBuffer.limit());
        }
        return null;
    }

3.MessageClientExt主要是這個(gè)對(duì)象惹的禍

老的版本mq就是一個(gè) MessageExt

我們?cè)賮?lái)看看這個(gè)對(duì)象為什么惹禍了

public class MessageClientExt extends MessageExt {
    public String getOffsetMsgId() {
        return super.getMsgId();
    }
    public void setOffsetMsgId(String offsetMsgId) {
        super.setMsgId(offsetMsgId);
    }
    /**
    * 沒(méi)錯(cuò)就是這里，默認(rèn)拿的是屬性里的UNIQ_KEY
    **/
    @Override
    public String getMsgId() {
        String uniqID = MessageClientIDSetter.getUniqID(this);
        if (uniqID == null) {
            return this.getOffsetMsgId();
        } else {
            return uniqID;
        }
    }
    public void setMsgId(String msgId) {
        //DO NOTHING
        //MessageClientIDSetter.setUniqID(this);
    }
}

所以知道了問(wèn)題，就好解決了

我是在顯示類(lèi)里面做相應(yīng)的處理，貼下我的處理方式，我是在 MessageView.class里處理

 public static MessageView fromMessageExt(MessageExt messageExt) {
        MessageView messageView = new MessageView();
        BeanUtils.copyProperties(messageExt, messageView);
        if (messageExt.getBody() != null) {
            messageView.setMessageBody(new String(messageExt.getBody(), Charsets.UTF_8));
        }
        //主要是這里判斷下，是否是這個(gè)類(lèi)，是就把原來(lái)的msgId拿出來(lái)
        if(messageExt instanceof MessageClientExt){
            MessageClientExt ext =   (MessageClientExt) messageExt;
            messageView.setMsgId(ext.getOffsetMsgId());
        }
        return messageView;
    }