RocketMQ之msgId查询逻辑

 @Override
  public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    SendMessageContext mqtraceContext = null;
    switch (request.getCode()) {
    case RequestCode.CONSUMER_SEND_MSG_BACK:
      return this.consumerSendMsgBack(ctx, request);
    default:
      SendMessageRequestHeader requestHeader = parseRequestHeader(request);
      if (requestHeader==null) {
        return null;
      }
      // 消息轨迹：记录到达 broker 的消息
      mqtraceContext = buildMsgContext(ctx, requestHeader);
      this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
      //处理生产者发送来的消息
      final RemotingCommand response = this.sendMessage(ctx, request, mqtraceContext, requestHeader);
      // 消息轨迹：记录发送成功的消息
      this.executeSendMessageHookAfter(response, mqtraceContext);
      return response;
    }
  }

private RemotingCommand sendMessage(final ChannelHandlerContext ctx, //
            final RemotingCommand request,//
            final SendMessageContext mqtraceContext,//
            final SendMessageRequestHeader requestHeader) throws RemotingCommandException {

     //省略代码
     MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
     //处理消息
     PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
     //省略代码
}

public PutMessageResult putMessage(MessageExtBrokerInner msg) {
     //省略非讲解代码
     PutMessageResult result = this.commitLog.putMessage(msg);
    //省略非讲解代码
}

public PutMessageResult putMessage(final MessageExtBrokerInner msg) {

  //省略非讲解代码
  result = mapedFile.appendMessage(msg, this.appendMessageCallback);
  //省略非讲解代码
}

public AppendMessageResult appendMessage(final Object msg, final AppendMessageCallback cb) {
        assert msg != null;
        assert cb != null;

        int currentPos = this.wrotePostion.get();

        // 表示有空余空间
        if (currentPos < this.fileSize) {
            ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
            byteBuffer.position(currentPos);
            //往MapedBuffer追加消息
            AppendMessageResult result =
                    cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, msg);
            this.wrotePostion.addAndGet(result.getWroteBytes());
            this.storeTimestamp = result.getStoreTimestamp();
            return result;
        }
}

public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer,
                final int maxBlank, final Object msg) {
            // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
            MessageExtBrokerInner msgInner = (MessageExtBrokerInner) msg;
            // PHY OFFSET
            long wroteOffset = fileFromOffset + byteBuffer.position();
            //msgId生成
            String msgId =
                    MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(),
                        wroteOffset);
            //省略非讲解代码
}

public MessageExt viewMessage(String msgId) throws RemotingException, MQBrokerException,
            InterruptedException, MQClientException {
        try {
            //解析出broker地址和phyoffset
            MessageId messageId = MessageDecoder.decodeMessageId(msgId);
            //调用另外一个重载viewMessage重载方法得到消息
            return this.mQClientFactory.getMQClientAPIImpl().viewMessage(
                RemotingUtil.socketAddress2String(messageId.getAddress()), messageId.getOffset(), 1000 * 3);
        }
        catch (UnknownHostException e) {
            throw new MQClientException("message id illegal", e);
        }
    }

public MessageExt viewMessage(final String addr, final long phyoffset, final long timeoutMillis)
            throws RemotingException, MQBrokerException, InterruptedException {
        ViewMessageRequestHeader requestHeader = new ViewMessageRequestHeader();
        requestHeader.setOffset(phyoffset);
        RemotingCommand request =
                RemotingCommand.createRequestCommand(RequestCode.VIEW_MESSAGE_BY_ID, requestHeader);
        //与broker建立链接,然后把phyoffset作为请求参数,即可得到消息
        RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
        case ResponseCode.SUCCESS: {
            ByteBuffer byteBuffer = ByteBuffer.wrap(response.getBody());
            MessageExt messageExt = MessageDecoder.decode(byteBuffer);
            if (!UtilAll.isBlank(projectGroupPrefix)) {
                messageExt.setTopic(VirtualEnvUtil.clearProjectGroup(messageExt.getTopic(),
                    projectGroupPrefix));
            }
            return messageExt;
        }
        default:
            break;
        }
        throw new MQBrokerException(response.getCode(), response.getRemark());
    }

@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
            throws RemotingCommandException {
        switch (request.getCode()) {
        case RequestCode.QUERY_MESSAGE:
            //按照Message Key查询消息
            return this.queryMessage(ctx, request);
        case RequestCode.VIEW_MESSAGE_BY_ID:
            //按照MessageId查询消息
            return this.viewMessageById(ctx, request);
        default:
            break;
        }

        return null;
}

public RemotingCommand viewMessageById(ChannelHandlerContext ctx, RemotingCommand request)
            throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final ViewMessageRequestHeader requestHeader =
                (ViewMessageRequestHeader) request.decodeCommandCustomHeader(ViewMessageRequestHeader.class);
        //省略非讲解代码
        final SelectMapedBufferResult selectMapedBufferResult =
                this.brokerController.getMessageStore().selectOneMessageByOffset(requestHeader.getOffset());
        //省略非讲解代码
}

@Override
public SelectMapedBufferResult selectOneMessageByOffset(long commitLogOffset) {
        SelectMapedBufferResult sbr = this.commitLog.getMessage(commitLogOffset, 4);
        if (null != sbr) {
            try {
                // 1 TOTALSIZE
                int size = sbr.getByteBuffer().getInt();
                return this.commitLog.getMessage(commitLogOffset, size);
            }
            finally {
                sbr.release();
            }
        }
        return null;
}