Java的RocketMQ之消息存储和查询原理详解

public static void main(String[] args) throws Exception {
    MQProducer producer = getProducer();
    for (int i = 0;i<10;i++){
        Message message = new Message();
        message.setTopic("topic"+i);
        message.setBody(("清幽之地的博客").getBytes());
        SendResult sendResult = producer.send(message);
    }
    producer.shutdown();
}

public static ByteBuffer read(String path)throws Exception{
    File file = new File(path);
    FileInputStream fin = new FileInputStream(file);
    byte[] bytes = new byte[(int)file.length()];
    fin.read(bytes);
    ByteBuffer buffer = ByteBuffer.wrap(bytes);
    return buffer;
}

/**
 * commitlog 文件解析
 * @param byteBuffer
 * @return
 * @throws Exception
 */
public static MessageExt decodeCommitLog(ByteBuffer byteBuffer)throws Exception {
	MessageExt msgExt = new MessageExt();
	// 1 TOTALSIZE
	int storeSize = byteBuffer.getInt();
	msgExt.setStoreSize(storeSize);
	if (storeSize<=0){
	    return null;
	}
	// 2 MAGICCODE
	byteBuffer.getInt();
	// 3 BODYCRC
	int bodyCRC = byteBuffer.getInt();
	msgExt.setBodyCRC(bodyCRC);
	// 4 QUEUEID
	int queueId = byteBuffer.getInt();
	msgExt.setQueueId(queueId);
	// 5 FLAG
	int flag = byteBuffer.getInt();
	msgExt.setFlag(flag);
	// 6 QUEUEOFFSET
	long queueOffset = byteBuffer.getLong();
	msgExt.setQueueOffset(queueOffset);
	// 7 PHYSICALOFFSET
	long physicOffset = byteBuffer.getLong();
	msgExt.setCommitLogOffset(physicOffset);
	// 8 SYSFLAG
	int sysFlag = byteBuffer.getInt();
	msgExt.setSysFlag(sysFlag);
	// 9 BORNTIMESTAMP
	long bornTimeStamp = byteBuffer.getLong();
	msgExt.setBornTimestamp(bornTimeStamp);
	// 10 BORNHOST
	int bornhostIPLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 : 16;
	byte[] bornHost = new byte[bornhostIPLength];
	byteBuffer.get(bornHost, 0, bornhostIPLength);
	int port = byteBuffer.getInt();
	msgExt.setBornHost(new InetSocketAddress(InetAddress.getByAddress(bornHost), port));
	// 11 STORETIMESTAMP
	long storeTimestamp = byteBuffer.getLong();
	msgExt.setStoreTimestamp(storeTimestamp);
	// 12 STOREHOST
	int storehostIPLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 : 16;
	byte[] storeHost = new byte[storehostIPLength];
	byteBuffer.get(storeHost, 0, storehostIPLength);
	port = byteBuffer.getInt();
	msgExt.setStoreHost(new InetSocketAddress(InetAddress.getByAddress(storeHost), port));
	// 13 RECONSUMETIMES
	int reconsumeTimes = byteBuffer.getInt();
	msgExt.setReconsumeTimes(reconsumeTimes);
	// 14 Prepared Transaction Offset
	long preparedTransactionOffset = byteBuffer.getLong();
	msgExt.setPreparedTransactionOffset(preparedTransactionOffset);
	// 15 BODY
	int bodyLen = byteBuffer.getInt();
	if (bodyLen > 0) {
	    byte[] body = new byte[bodyLen];
	    byteBuffer.get(body);
	    msgExt.setBody(body);
	}
	// 16 TOPIC
	byte topicLen = byteBuffer.get();
	byte[] topic = new byte[(int) topicLen];
	byteBuffer.get(topic);
	msgExt.setTopic(new String(topic, CHARSET_UTF8));
	// 17 properties
	short propertiesLength = byteBuffer.getShort();
	if (propertiesLength > 0) {
	    byte[] properties = new byte[propertiesLength];
	    byteBuffer.get(properties);
	    String propertiesString = new String(properties, CHARSET_UTF8);
	    Map<String, String> map = string2messageProperties(propertiesString);
	}
	int msgIDLength = storehostIPLength + 4 + 8;
	ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
	String msgId = createMessageId(byteBufferMsgId, msgExt.getStoreHostBytes(), msgExt.getCommitLogOffset());
	msgExt.setMsgId(msgId);
	return msgExt;
}

public static void main(String[] args) throws Exception {
    String filePath = "C:\\Users\\shiqizhen\\store\\commitlog\\00000000000000000000";
    ByteBuffer buffer = read(filePath);
    List<MessageExt> messageList = new ArrayList<>();
    while (true){
        MessageExt message = decodeCommitLog(buffer);
        if (message==null){
            break;
        }
        messageList.add(message);
    }
    for (MessageExt ms:messageList) {
        System.out.println("主题:"+ms.getTopic()+" 消息:"+
            new String(ms.getBody())+"队列ID:"+ms.getQueueId()+" 存储地址:"+ms.getStoreHost());
    }
}

public MessageQueue selectOneMessageQueue() {
    int index = this.sendWhichQueue.getAndIncrement();
    int pos = Math.abs(index) % this.messageQueueList.size();
    if (pos < 0)
        pos = 0;
    return this.messageQueueList.get(pos);
}

public static void main(String[] args)throws Exception {
    String path = "C:\\Users\\shiqizhen\\store\\consumequeue\\order\\0\\00000000000000000000";
    ByteBuffer buffer = read(path);
    while (true){
        long offset = buffer.getLong();
        long size = buffer.getInt();
        long code = buffer.getLong();
        if (size==0){
            break;
        }
        System.out.println("消息长度:"+size+" 消息偏移量:" +offset);
    }
    System.out.println("--------------------------");
}

public static MessageExt getMessageByOffset(ByteBuffer commitLog,long offset,int size) throws Exception {
    ByteBuffer slice = commitLog.slice();
    slice.position((int)offset);
    slice.limit((int) (offset+size));
    MessageExt message = CommitLogTest.decodeCommitLog(slice);
    return message;
}

public static void main(String[] args) throws Exception {
	//consumerqueue根目录
	String consumerPath = "C:\\Users\\shiqizhen\\store\\consumequeue";
	//commitlog目录
	String commitLogPath = "C:\\Users\\shiqizhen\\store\\commitlog\\00000000000000000000";
	//读取commitlog文件内容
	ByteBuffer commitLogBuffer = CommitLogTest.read(commitLogPath);
	//遍历consumerqueue目录下的所有文件
	File file = new File(consumerPath);
	File[] files = file.listFiles();
	for (File f:files) {
		if (f.isDirectory()){
			File[] listFiles = f.listFiles();
			for (File queuePath:listFiles) {
				String path = queuePath+"/00000000000000000000";
				//读取consumerqueue文件内容
				ByteBuffer buffer = CommitLogTest.read(path);
				while (true){
					//读取消息偏移量和消息长度
					long offset = (int) buffer.getLong();
					int size = buffer.getInt();
					long code = buffer.getLong();
					if (size==0){
						break;
					}
					//根据偏移量和消息长度，在commitloh文件中读取消息内容
					MessageExt message = getMessageByOffset(commitLogBuffer,offset,size);
					if (message!=null){
						System.out.println("消息主题:"+message.getTopic()+" MessageQueue:"+
							message.getQueueId()+" 消息体:"+new String(message.getBody()));
					}
				}
			}
		}
	}
}

public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
    //ConsumerQueue文件大小
    int mappedFileSize = this.mappedFileSize;	
    //根据消费进度，找到在consumerqueue文件里的偏移量
    long offset = startIndex * CQ_STORE_UNIT_SIZE;
    if (offset >= this.getMinLogicOffset()) {
        //返回ConsumerQueue映射文件
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
        if (mappedFile != null) {
            //返回文件里的某一块内容
            SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize));
            return result;
        }
    }
    return null;
}

public SelectMappedBufferResult getMessage(final long offset, final int size) {
    //commitlog文件大小
    int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
    //根据消息偏移量，定位到具体的commitlog文件
    MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
    if (mappedFile != null) {
        //根据消息偏移量和长度，获取消息内容
        int pos = (int) (offset % mappedFileSize);
        return mappedFile.selectMappedBuffer(pos, size);
    }
    return null;
}

/**
 * 创建消息ID
 * @param input     
 * @param addr      Broker服务器地址
 * @param offset    正在存储的消息，在Commitlog中的偏移量
 * @return
 */
public static String createMessageId(final ByteBuffer input, final ByteBuffer addr, final long offset) {
    input.flip();
    int msgIDLength = addr.limit() == 8 ? 16 : 28;
    input.limit(msgIDLength);
    input.put(addr);
    input.putLong(offset);
    return UtilAll.bytes2string(input.array());
}

public static MessageId decodeMessageId(final String msgId) throws Exception {
    SocketAddress address;
    long offset;
    int ipLength = msgId.length() == 32 ? 4 * 2 : 16 * 2;
    byte[] ip = UtilAll.string2bytes(msgId.substring(0, ipLength));
    byte[] port = UtilAll.string2bytes(msgId.substring(ipLength, ipLength + 8));
    ByteBuffer bb = ByteBuffer.wrap(port);
    int portInt = bb.getInt(0);
    //解析出来Broker地址
    address = new InetSocketAddress(InetAddress.getByAddress(ip), portInt);
    //偏移量
    byte[] data = UtilAll.string2bytes(msgId.substring(ipLength + 8, ipLength + 8 + 16));
    bb = ByteBuffer.wrap(data);
    offset = bb.getLong(0);
    return new MessageId(address, offset);
}

public static void main(String[] args) throws Exception {
    //index索引文件的路径
    String path = "C:\\Users\\shiqizhen\\store\\index\\20200506224547616";
    ByteBuffer buffer = CommitLogTest.read(path);
    //该索引文件中包含消息的最小存储时间
    long beginTimestamp = buffer.getLong();
    //该索引文件中包含消息的最大存储时间
    long endTimestamp = buffer.getLong();
    //该索引文件中包含消息的最大物理偏移量(commitlog文件偏移量)
    long beginPhyOffset = buffer.getLong();
    //该索引文件中包含消息的最大物理偏移量(commitlog文件偏移量)
    long endPhyOffset = buffer.getLong();
    //hashslot个数
    int hashSlotCount = buffer.getInt();
    //Index条目列表当前已使用的个数
    int indexCount = buffer.getInt();
    //500万个hash槽，每个槽占4个字节，存储的是index索引
    for (int i=0;i<5000000;i++){
        buffer.getInt();
    }
    //2000万个index条目
    for (int j=0;j<20000000;j++){
        //消息key的hashcode
        int hashcode = buffer.getInt();
        //消息对应的偏移量
        long offset = buffer.getLong();
        //消息存储时间和第一条消息的差值
        int timedif = buffer.getInt();
        //该条目的上一条记录的index索引
        int pre_no = buffer.getInt();
    }
    System.out.println(buffer.position()==buffer.capacity());
}

public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
    if (this.indexHeader.getIndexCount() < this.indexNum) {
        //计算key的hash
        int keyHash = indexKeyHashMethod(key);
        //计算hash槽的坐标
        int slotPos = keyHash % this.hashSlotNum;
        int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
        //计算时间差值
        long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();
        timeDiff = timeDiff / 1000;
        //计算INDEX条目的起始偏移量
        int absIndexPos =
            IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                + this.indexHeader.getIndexCount() * indexSize;
        //依次写入hashcode、消息偏移量、时间戳、hash槽的值
        this.mappedByteBuffer.putInt(absIndexPos, keyHash);
        this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
        this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
        this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
        //将当前INDEX中包含的条目数量写入HASH槽
        this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
        return true;
    }
    return false;
}

adminImpl.queryMessageByUniqKey("order", "FD88E3AB24F6980059FDC9C3620464741BCC18B4AAC220FDFE890007");