自己动手实现一个RocketMQ文件存储系统

　　前言
　　RocketMQ作为一款优秀的开源消息中间件，很多java开发者都在使用并研究里面的源码。目前网上有很多关于RocketMQ源代码的文章，但是很多文章只是从框架开发者的的角度分析源码，没有从技术实现本质进行剖析。因此很多源码学习者在读完后还是一知半解，当自己想动手写的时候不知要用到哪种技术，无从着手。笔者基于对RocketMQ的文件存储研究，结合开发者常见的技术，自己动手实现了一个简化版本的RocketMQ文件系统，希望能抽丝剥茧，帮助开发者从本质上理解RocketMQ文件存储的原理，起到抛砖引玉，举一反三的作用。
　　RocketMQ逻辑存储结构
　　本文适合对RocketMQ的文件存储原理有一定的了解，熟悉java NIO,希望了解RocketMQ是如何通过java NIO实现的读者。以下代码部分：
　　1.手动生成10个消息，并创建commitLog文件，consumeQueue,indexFile文件package org.apache.rocketmq.test.smoke;  import com.alibaba.fastjson.JSON; import org.apache.rocketmq.test.smoke.model.ConsumerQueueData; import org.apache.rocketmq.test.smoke.model.IndexFileHeaderData; import org.apache.rocketmq.test.smoke.model.IndexFileItemData;  import java.io.IOException; import java.net.URI; import java.nio.ByteBuffer; import java.nio.MappedByteBuffer; import java.nio.channels.FileChannel; import java.nio.charset.StandardCharsets; import java.nio.file.Paths; import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.List; import java.util.Random; import java.util.UUID;  public class CommitLogWriteTest {     private static Long commitLogOffset = 0L;//8byte(commitlog offset)     private static Long lastTotalSize = 0L;     private static Long currentTotalSize = 0L;     private static List consumerQueueDatas = new ArrayList<>();     private static List indexFileItemDatas = new ArrayList<>();     private static int MESSAGE_COUNT = 10;      public static void main(String[] args) throws IOException {         createCommitLog();         createConsumerQueue();         createIndexFile();     }      private static void createCommitLog() throws IOException {         FileChannel fileChannel = FileChannel.open(Paths.get(URI.create(＂file:/c:/123/commitLog.txt＂)),                 StandardOpenOption.WRITE, StandardOpenOption.READ);         MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 409600);         fileChannel.close();         Random random = new Random();          int count = 0;         for (int i = 0; i < MESSAGE_COUNT; i++) {             long commitLogOffset = lastTotalSize;              String topic = ＂Topic-test＂;             String msgId = UUID.randomUUID().toString();             String msgBody = ＂消息内容＂ + ＂msgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsg＂.substring(0, random.nextInt(48));//             long queueOffset = i;//索引偏移量             String transactionId = UUID.randomUUID().toString();            /* 数据格式，位置固定          int totalSize;//消息长度          String msgId;          String topic;          long queueOffset;//索引偏移量          long bodySize;//消息长度          byte[] body;//消息内容          String transactionId;          long commitLogOffset;//从第一个文件开始算的偏移量           */              int totalSize = 8 //totalSize长度                     + 64  //msgId长度                     + 64 //topic长度                     + 8 //索引偏移量长度                     + 8 //消息长度长度                     + msgBody.getBytes(StandardCharsets.UTF_8).length //消息内容长度                     + 64  //transactionId长度                     + 64  //commitLogOffset长度;                     ;              ByteBuffer b = ByteBuffer.allocate(totalSize);             // //如果3个消息长度分别是100，200，350，则偏移量分别是0,100,300             mappedByteBuffer.position(Integer.valueOf(commitLogOffset + ＂＂));              b.putLong(totalSize);//totalSize             b.put(getBytes(msgId, 64));//msgId             b.put(getBytes(topic, 64));//topic,定长64             b.putLong(queueOffset);//索引偏移量             b.putLong(msgBody.getBytes(StandardCharsets.UTF_8).length);//bodySize             b.put(msgBody.getBytes(StandardCharsets.UTF_8));//body             b.put(getBytes(transactionId, 64));             b.putLong(commitLogOffset);//bodySize             b.flip();             mappedByteBuffer.put(b);              lastTotalSize = totalSize + lastTotalSize;              System.out.println(＂写入消息，第:＂ + i + ＂次＂);              System.out.println(＂totalSize:＂ + totalSize);             System.out.println(＂msgId:＂ + msgId);             System.out.println(＂topic:＂ + topic);             System.out.println(＂msgBody:＂ + msgBody);             System.out.println(＂transactionId:＂ + transactionId);             System.out.println(＂commitLogOffset:＂ + commitLogOffset);              ConsumerQueueData consumerQueueData = new ConsumerQueueData();             consumerQueueData.setOffset(commitLogOffset);             consumerQueueData.setMsgLength(totalSize);             consumerQueueData.setTagCode(100L);             //准备生成consumeQueue文件             consumerQueueDatas.add(consumerQueueData);              IndexFileItemData indexFileItemData = new IndexFileItemData();             indexFileItemData.setKeyHash(msgId.hashCode());             indexFileItemData.setMessageId(msgId);             indexFileItemData.setPhyOffset(commitLogOffset);             //准备生成indexFile文件             indexFileItemDatas.add(indexFileItemData);             mappedByteBuffer.force();             count++;         }          System.out.println(＂commitLog数据保存完成,totalSize:＂ + count);      }       public static void createConsumerQueue() throws IOException {         FileChannel fileChannel = FileChannel.open(Paths.get(URI.create(＂file:/c:/123/consumerQueue.txt＂)),                 StandardOpenOption.WRITE, StandardOpenOption.READ);         MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 4096);         fileChannel.close();         int count = 0;         for (int i = 0; i < consumerQueueDatas.size(); i++) {             ConsumerQueueData consumerQueueData = consumerQueueDatas.get(i);             //指定写入位置             mappedByteBuffer.position(i * 20);             mappedByteBuffer.putLong(consumerQueueData.getOffset());//8byte(commitlog offset)             mappedByteBuffer.putInt(consumerQueueData.getMsgLength());//4byte （msgLength)             mappedByteBuffer.putLong(consumerQueueData.getTagCode());//8byte (tagCode)              count++;             System.out.println(＂consumerQueue数据写入完成:＂ + JSON.toJSONString(consumerQueueData));             mappedByteBuffer.force();          }         System.out.println(＂ConsumerQueue数据保存完成count:＂ + count);      }        public static void createIndexFile() throws IOException {         //文件场创建时间，在写第一条消息的时候创建         FileChannel fileChannel = FileChannel.open(Paths.get(URI.create(＂file:/c:/123/index.txt＂)),                 StandardOpenOption.WRITE, StandardOpenOption.READ);         MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 409600);         ByteBuffer headerByteBuffer = mappedByteBuffer.slice();         long firstDataTime = System.currentTimeMillis();          fileChannel.close();          //开始写hash槽，从头部后写入         /*  已经填充有index的slot数量           （并不是每个slot槽下都挂载有index索引单元，这 里统计的是所有挂载了index索引单元的slot槽的数量，hash冲突）*/         int hashSlotCount = 0;          /* 已该indexFile中包含的索引单元个数（统计出当前indexFile中所有slot槽下挂载的所有index索引单元的数量之和），         如果没有hash冲突,hashSlotCount = indexCount*/         int indexCount = 0;          //假设建立100个槽位（总长度400）         int soltNum = 100;          for (int i = 0; i < MESSAGE_COUNT; i++) {             IndexFileItemData indexFileItemData = indexFileItemDatas.get(i);             int keyHash = indexFileItemData.getKeyHash();              //取模，计算第几个槽位             int slotPos = keyHash % 100 > 0?keyHash % 100:-1*(keyHash % 100);              // slot存放第几条数据的的位置（字节数组位置）             int absSlotPos = 40 + slotPos * 4;              // 存储实际数据的位置（字节数组位置）             int absIndexPos =                     40 + soltNum * 4                             + indexCount * 20;                //将hash槽的值设置为indexCount，建立索引,即第n条消息保存在XX位置             mappedByteBuffer.putInt(absSlotPos, indexCount);              //写入数据             mappedByteBuffer.putInt(absIndexPos,indexFileItemData.getKeyHash());//4byte msg hashcode             mappedByteBuffer.putLong(absIndexPos+4,indexFileItemData.getPhyOffset());//8byte pyhoffset             mappedByteBuffer.putInt(absIndexPos+4+8,Integer.valueOf((System.currentTimeMillis()- firstDataTime)+＂＂));//4byte (timeDiff)             mappedByteBuffer.putInt(absIndexPos+4+8+4,0);//4byte (preIndex),暂置0，暂不考虑hash冲突的情况               //模拟最后一个文件，写入header             if (i == 0) {                 //该indexFile中第一条消息的存储时间                 headerByteBuffer.putLong(0, firstDataTime);                 //该indexFile种第一条消息在commitlog种的偏移量commitlog offset                 mappedByteBuffer.putLong(16, indexFileItemData.getPhyOffset());             }             //模拟最后一个文件，写入header             if (i == 99) {                 //该indexFile种最后一条消息存储时间                 headerByteBuffer.putLong(8, System.currentTimeMillis());                 //该indexFile中最后一条消息在commitlog中的偏移量commitlog offset                 headerByteBuffer.putLong(24, indexFileItemData.getPhyOffset());             }             //已经填充有index的slot数量             headerByteBuffer.putInt(32, hashSlotCount+1);             //该indexFile中包含的索引单元个数             headerByteBuffer.putInt(36, indexCount+1);             mappedByteBuffer.force();             System.out.println(＂msgId:＂+indexFileItemData.getMessageId()+＂,keyHash:＂+keyHash+＂,保存槽位为＂+slotPos+＂的数据,absSlotPos=＂+absSlotPos+＂,值index=＂+indexCount+＂,绝对位置:＂+absIndexPos+＂,commit-phyOffset:＂+indexFileItemData.getPhyOffset());              indexCount ++;             hashSlotCount++;          }      }        //将变长字符串定长byte[]，方便读取     private static byte[] getBytes(String s, int length) {         int fixLength = length - s.getBytes().length;         if (s.getBytes().length < length) {             byte[] S_bytes = new byte[length];             System.arraycopy(s.getBytes(), 0, S_bytes, 0, s.getBytes().length);             for (int x = length - fixLength; x < length; x++) {                 S_bytes[x] = 0x00;             }             return S_bytes;         }         return s.getBytes(StandardCharsets.UTF_8);     }  }
　　运行结果：写入消息，第:0次 totalSize:322 msgId:61b5c500-f7f5-4bfe-beb1-50a8148534c0 topic:Topic-test msgBody:消息内容msgmsgmsgmsgmsgmsgmsgmsgmsgmsg transactionId:9453ba39-3982-40e9-926d-47b51d360590 commitLogOffset:0 写入消息，第:1次 totalSize:306 msgId:d0fbf80f-223b-4721-a43e-518b152decc2 topic:Topic-test msgBody:消息内容msgmsgmsgmsgms transactionId:e2ef1652-58fa-4849-bf74-885c7e5db9e3 commitLogOffset:322 写入消息，第:2次 totalSize:307 msgId:199053e3-e616-4611-ab0d-e5c7af4549a9 topic:Topic-test msgBody:消息内容msgmsgmsgmsgmsg transactionId:33d21abe-0d8e-4c0e-9c78-f415daefd767 commitLogOffset:628 写入消息，第:3次 totalSize:339 msgId:8e799d8e-3290-4f6b-ab5d-289153446994 topic:Topic-test msgBody:消息内容msgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgmsgms transactionId:98b46b96-cc88-4969-a56f-282d25799085 commitLogOffset:935 写入消息，第:4次 totalSize:320 msgId:8b78474f-b28a-4442-99a0-6f7883f0302b topic:Topic-test msgBody:消息内容msgmsgmsgmsgmsgmsgmsgmsgmsgm transactionId:5c0ff6fe-aea3-40c3-8647-6f4bdd797a78 commitLogOffset:1274 写入消息，第:5次 totalSize:312 msgId:b33c6f31-cc96-462b-b095-99410459082c topic:Topic-test msgBody:消息内容msgmsgmsgmsgmsgmsgms transactionId:57420047-2539-43fa-a3f2-b2f55c7b059c commitLogOffset:1594 写入消息，第:6次 totalSize:324 msgId:d0a6803f-8555-418e-988a-b3b9a70d14f0 topic:Topic-test msgBody:消息内容msgmsgmsgmsgmsgmsgmsgmsgmsgmsgms transactionId:29601335-3fcd-4193-b14f-140bbaf409a4 commitLogOffset:1906 写入消息，第:7次 totalSize:293 msgId:91151ec5-e76b-4560-90b7-ab77f9d04c9a topic:Topic-test msgBody:消息内容m transactionId:291e54de-2ebe-41b1-b974-e81a2e9f1370 commitLogOffset:2230 写入消息，第:8次 totalSize:323 msgId:eb21df35-b4dc-43aa-8604-e9a103f25a7b topic:Topic-test msgBody:消息内容msgmsgmsgmsgmsgmsgmsgmsgmsgmsgm transactionId:3005a39d-b8cb-4138-ae05-34b65fc135a2 commitLogOffset:2523 写入消息，第:9次 totalSize:296 msgId:abcda364-8fc3-4d18-ae5d-1d7a8ffd0929 topic:Topic-test msgBody:消息内容msgm transactionId:d42733b5-3911-4f0a-b1db-11eb45a30345 commitLogOffset:2846 commitLog数据保存完成,totalSize:10  创建consumerQueue文件开始 consumerQueue数据写入完成:{＂msgLength＂:322,＂offset＂:0,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:306,＂offset＂:322,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:307,＂offset＂:628,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:339,＂offset＂:935,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:320,＂offset＂:1274,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:312,＂offset＂:1594,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:324,＂offset＂:1906,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:293,＂offset＂:2230,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:323,＂offset＂:2523,＂tagCode＂:100} consumerQueue数据写入完成:{＂msgLength＂:296,＂offset＂:2846,＂tagCode＂:100} ConsumerQueue数据保存完成count:10  创建索引文件开始 msgId:61b5c500-f7f5-4bfe-beb1-50a8148534c0,keyHash:249765627,保存槽位为27的数据,absSlotPos=148,值index=0,绝对位置:440,commit-phyOffset:0 msgId:d0fbf80f-223b-4721-a43e-518b152decc2,keyHash:1587335015,保存槽位为15的数据,absSlotPos=100,值index=1,绝对位置:460,commit-phyOffset:322 msgId:199053e3-e616-4611-ab0d-e5c7af4549a9,keyHash:791210473,保存槽位为73的数据,absSlotPos=332,值index=2,绝对位置:480,commit-phyOffset:628 msgId:8e799d8e-3290-4f6b-ab5d-289153446994,keyHash:1460275929,保存槽位为29的数据,absSlotPos=156,值index=3,绝对位置:500,commit-phyOffset:935 msgId:8b78474f-b28a-4442-99a0-6f7883f0302b,keyHash:1174005465,保存槽位为65的数据,absSlotPos=300,值index=4,绝对位置:520,commit-phyOffset:1274 msgId:b33c6f31-cc96-462b-b095-99410459082c,keyHash:-1695757800,保存槽位为0的数据,absSlotPos=40,值index=5,绝对位置:540,commit-phyOffset:1594 msgId:d0a6803f-8555-418e-988a-b3b9a70d14f0,keyHash:1334295408,保存槽位为8的数据,absSlotPos=72,值index=6,绝对位置:560,commit-phyOffset:1906 msgId:91151ec5-e76b-4560-90b7-ab77f9d04c9a,keyHash:1287318090,保存槽位为90的数据,absSlotPos=400,值index=7,绝对位置:580,commit-phyOffset:2230 msgId:eb21df35-b4dc-43aa-8604-e9a103f25a7b,keyHash:239865974,保存槽位为74的数据,absSlotPos=336,值index=8,绝对位置:600,commit-phyOffset:2523 msgId:abcda364-8fc3-4d18-ae5d-1d7a8ffd0929,keyHash:-1173357775,保存槽位为75的数据,absSlotPos=340,值index=9,绝对位置:620,commit-phyOffset:2846
　　2.读取consumeQueue文件，并根据offset从commitLog读取一条完整的消息package org.apache.rocketmq.test.smoke;  import java.io.IOException; import java.net.URI; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.MappedByteBuffer; import java.nio.channels.FileChannel; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import java.nio.file.Paths; import java.nio.file.StandardOpenOption;  public class ConsumeQueueMessageReadTest {     public static MappedByteBuffer mappedByteBuffer = null;     private static int MESSAGE_COUNT = 10;      public static void main(String[] args) throws IOException {         FileChannel fileChannel = FileChannel.open(Paths.get(URI.create(＂file:/c:/123/consumerQueue.txt＂)),                 StandardOpenOption.WRITE, StandardOpenOption.READ);         MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 4096);         fileChannel.close();         mappedByteBuffer.position(0);         //根据索引下标读取索引，实际情况是用户消费的最新点位，存在在broker的偏移量文件中         int index = 0 ;         for(int i =index;i 0?keyHash % 100:-1*(keyHash % 100);         System.out.println(＂messageId:＂+messageId+ ＂,取模为:＂+slotPos);          // slot在文件中的字节数组位置         int absSlotPos = 40 + slotPos * 4;         System.out.println(＂哈希槽的字节数组位置:(40+＂+slotPos+＂*4)=＂+absSlotPos);           //获取hash槽上存取的件索引,第几个文件         int index =indexMappedByteBuffer.getInt(absSlotPos);          //计算数据需要存储的偏移量         int absIndexPos =                 40 + 100 * 4                         + index * 20;          System.out.println(＂第几个文件index=＂+index+＂，实际存储数据的字节数组位置:(40 + 100 * 4+index *20)=＂+absIndexPos);          long keyHash1 = indexMappedByteBuffer.getInt(absIndexPos);         long pyhOffset = indexMappedByteBuffer.getLong(absIndexPos+4);         int timeDiff = indexMappedByteBuffer.getInt(absIndexPos+4+8);         int preIndexNo = indexMappedByteBuffer.getInt(absIndexPos+4+8+4);           System.out.println(＂从index获取到的commitLog偏移量为:＂+pyhOffset);         System.out.println(＂＂);          readCommitLogByOffset((int)pyhOffset);      }       public static MappedByteBuffer initFileChannel() throws IOException {         if(mappedByteBuffer == null){           FileChannel  commitLogfileChannel = FileChannel.open(Paths.get(URI.create(＂file:/c:/123/commitLog.txt＂)),                     StandardOpenOption.WRITE, StandardOpenOption.READ);               mappedByteBuffer = commitLogfileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 409600);             commitLogfileChannel.close();         }          return mappedByteBuffer;      }       /*     *     * 根据偏移量读取CcommitLog     * */     public static void readCommitLogByOffset(int offset) throws IOException {           /* 存放顺序，读时候保持顺序一致             b.putLong(totalSize);//totalSize             b.put(getBytes(msgId, 64));//msgId             b.put(getBytes(topic, 64));//topic,定长64             b.putLong(queueOffset);//索引偏移量             b.putLong(msgBody.getBytes(StandardCharsets.UTF_8).length);//bodySize             b.put(msgBody.getBytes(StandardCharsets.UTF_8));//body             b.put(getBytes(transactionId, 64));             b.putLong(commitLogOffset);//commitLogOffset         */         System.out.println(＂=================commitlog读取偏移量为＂+offset+＂的消息===================＂);          MappedByteBuffer  mappedByteBuffer = initFileChannel();         mappedByteBuffer.position(offset);           long totalSize = mappedByteBuffer.getLong();//消息长度          byte[] msgIdByte = new byte[64];//uuid 固定是64         mappedByteBuffer.get(msgIdByte);          byte[] topicByte = new byte[64];// 固定是64         mappedByteBuffer.get(topicByte);         long queueOffset = mappedByteBuffer.getLong();         Long bodySize = mappedByteBuffer.getLong();         int bSize = Integer.valueOf(bodySize+＂＂);         byte[] bodyByte = new  byte[bSize];//bodySize 长度不固定         mappedByteBuffer.get(bodyByte);         byte[] transactionIdByte = new byte[64];//uuid 固定是64         mappedByteBuffer.get(transactionIdByte);         long commitLogOffset = mappedByteBuffer.getLong();//偏移量         System.out.println(＂totalSize:＂+totalSize);         System.out.println(＂msgId:＂+new String(msgIdByte));         System.out.println(＂topic:＂+new String(topicByte));         System.out.println(＂queueOffset:＂+queueOffset);         System.out.println(＂bodySize:＂+bodySize);         System.out.println(＂body:＂+new String(bodyByte));         System.out.println(＂transactionId:＂+new String(transactionIdByte));         System.out.println(＂commitLogOffset:＂+commitLogOffset);      }        public static byte[] toByteArray(long number) {         byte length = Long.BYTES;         byte[] bytes = new byte[length];          for (byte i = 0; i < length; i++) {             bytes[length - 1 - i] = (byte) number;             number >>= 8;         }          return bytes;      }  }
　　运行结果：============get indexFile header=============== beginTimestampIndex:1669554286826 endTimestampIndex:1669552196010 beginPhyoffsetIndex:0 endPhyoffsetIndex:31259 hashSlotcountIndex:10 indexCountIndex:10  messageId:8b78474f-b28a-4442-99a0-6f7883f0302b,取模为:65 哈希槽的字节数组位置:(40+65*4)=300 第几个文件index=4，实际存储数据的字节数组位置:(40 + 100 * 4+index *20)=520 从index获取到的commitLog偏移量为:1274  =================commitlog读取偏移量为1274的消息=================== totalSize:320 msgId:8b78474f-b28a-4442-99a0-6f7883f0302b                             topic:Topic-test                                                       queueOffset:4 bodySize:40 body:消息内容msgmsgmsgmsgmsgmsgmsgmsgmsgm transactionId:5c0ff6fe-aea3-40c3-8647-6f4bdd797a78                             commitLogOffset:1274
　　本文基于java NIO实现了RocketMQ的文件系统的最精简的实现，希望能帮助相关开发人员了解文件系统底层的实现原理。欢迎一起交流讨论，不足的地方欢迎指正。