MetaQ源码分析——架构、配置、消息存储

系统管理员 2021-12-19 03:37 252阅读 0赞

# MetaQ总体架构 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70]

## MetaQ工程结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 1]

*  metamorphosis-client：生产者和消费者客户端
 *  metamorphosis-client-extension：扩展的客户端。用于将消费处理失败的消息存入notify(未提供),和使用meta作为log4j appender，可以透明地使用log4j API发送消息到meta
 *  metamorphosis-commons：客户端和服务端一些公用的东西
 *  metamorphosis-dashboard,：metaQ的Web信息展示
 *  metamorphosis-example: 客户端使用的例子
 *  metamorphosis-http-client: 使用http协议的客户端
 *  metamorphosis-server: 服务端工程
 *  metamorphosis-server-wrapper: 扩展的服务端，用于将其他插件集成到服务端，提供扩展功能，支持高可用的同步异步复制
 *  metamorphosis-storm-spout: 用于将meta消息接入到twitter storm集群做实时分析
 *  metamorphosis-tools: 提供服务端管理和操作的一些工具

工程依赖图大致如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 2]  
**源码分析也分为Server和Client进行**

--------------------

# MetaQ-Server源码分析 #

MetaQ分为Broker、Client、Zookeeper，系统结构图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 3]

## Broker ##

Broker主要围绕发送消息和消费消息的主线进行，对于Broker来说就是输入、输出流的处理  
在该主线下面，Broker主要分为如下几个模块：

*  网络传输模块
 *  消息存储模块
 *  消息统计模块
 *  事物模块

### 配置模块 ###

我们得了解Broker中的一个重要的类MetaConfig，MetaConfig是Broker配置加载器，通过MetaConfig可以获取到各模块相关的配置，所以MetaConfig是贯穿所有模块的类。MetaConfig实现MetaConfigMBean接口，该接口定义如下：  
![在这里插入图片描述][20190707123445343.png]

public interface MetaConfigMBean { 
        /** * Reload topics configuration */
        public void reload();
    
    
        /** 关闭分区 */
        public void closePartitions(String topic, int start, int end);
    
    
        /** 打开一个topic的所有分区 */
        public void openPartitions(String topic);
    
    }

MetaConfig实现上面这个接口  
![在这里插入图片描述][20190707123943638.png]

public class MetaConfig extends Config implements Serializable, MetaConfigMBean { 
        static final long serialVersionUID = -1L;
        private int brokerId = 0;
        private String dataPath = System.getProperty("user.home") + File.separator + "meta";
        private int serverPort = 8123;
        private int dashboardHttpPort = 8120;
        private String hostName;
        private int numPartitions = 1;
        private int unflushThreshold = 1000;
        private int unflushInterval = 10000;
        private int maxSegmentSize = 1 * 1024 * 1024 * 1024;
        private int maxTransferSize = 1024 * 1024;
    
        private List<String> topics = new ArrayList<String>();
    
    	......

MetaConfig注册到了MBeanServer上，所以可以通过JMX协议重新加载配置以及关闭和打开分区。

@Override
        public void reload() { 
            final File file = new File(this.path);
            if (file.lastModified() != this.lastModified) { 
                try { 
                    log.info("Reloading topics......");
                    final Ini conf = this.createIni(file);
                    MetaConfig.this.populateTopicsConfig(conf);
                    log.info("Reload topics successfully");
                }
                catch (final Exception e) { 
                    log.error("Reload config failed", e);
                }
            }
        }

*  Ini可以看做是处理配置的一个类
 *  调用reload之后，可以看到只对topic配置生效，而对于全局的配置不生效，只是重新加载topic的配置。

MetaConfig内置了一个通知机制，可以通过向MetaConfig注册监听器的方式关注相关配置的变化，监听器需实现PropertyChangeListener接口。

public void addPropertyChangeListener(final String propertyName, final PropertyChangeListener listener) { 
          this.propertyChangeSupport.addPropertyChangeListener(propertyName, listener);
    }
     
    public void removePropertyChangeListener(final PropertyChangeListener listener) { 
          this.propertyChangeSupport.removePropertyChangeListener(listener);
    }

目前MetaConfig发出的事件通知有三种：配置文件发生变化（configFileChecksum）、主题发生变化（topics，topicConfigMap）以及刷新存储的频率发生变化（unflushInterval），代码如下：

//configFileChecksum通知
    private Ini createIni(final File file) throws IOException, InvalidFileFormatException { 
          ……
          this.propertyChangeSupport.firePropertyChange("configFileChecksum", null, null);
          return conf;
    }
     
    public void setConfigFileChecksum(long configFileChecksum) { 
          this.configFileChecksum = configFileChecksum;
          this.propertyChangeSupport.firePropertyChange("configFileChecksum", null, null);
    }
     
    //topics、topicConfigMap和unflushInterval通知
    private void populateTopicsConfig(final Ini conf) { 
    ……
    if (!newTopicConfigMap.equals(this.topicConfigMap)) { 
              this.topics = newTopics;
              this.topicConfigMap = newTopicConfigMap;
              this.propertyChangeSupport.firePropertyChange("topics", null, null);
              this.propertyChangeSupport.firePropertyChange("topicConfigMap", null, null);
      }
      this.propertyChangeSupport.firePropertyChange("unflushInterval", null, null);

### 消息存储模块 ###

Broker的存储模块用于存储Client发送的等待被消费的消息，Broker采用文件存储的方式来存储消息，存储模块的类图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 4]

#### 1. MessageSet模块 ####

MessageSet代表的是一个消息集合，可能是一个文件或者文件的一部分，其定义如下：

public interface MessageSet { 
      public MessageSet slice(long offset, long limit) throws IOException; //获取一个消息集合
     
      public void write(GetCommand getCommand, SessionContext ctx);
     
      public long append(ByteBuffer buff) throws IOException; //存储一个消息，这时候还没有存储到磁盘，需要调用flush方法才能保证存储到磁盘
     
      public void flush() throws IOException; //提交到磁盘
     
      public void read(final ByteBuffer bf, long offset) throws IOException; //读取消息
     
      public void read(final ByteBuffer bf) throws IOException; //读取消息
     
      public long getMessageCount();//该集合的消息数量
    }

FileMessageSet实现了MessageSet接口和Closeable接口，实现Closeable接口主要是为了在文件关闭的时候确保文件通道关闭以及内容是否提交到磁盘。

public class FileMessageSet implements MessageSet, Closeable { 
    
        private final FileChannel channel;
        private final AtomicLong messageCount;
        private final AtomicLong sizeInBytes;
        private final AtomicLong highWaterMark; // 已经确保写入磁盘的水位
        private final long offset; // 镜像offset
        private boolean mutable; // 是否可变
    
        static final Log log = LogFactory.getLog(FileMessageSet.class);

关闭文件通道函数：

public void close() throws IOException { 
          if (!this.channel.isOpen()) { 
              return;
          }
          //保证在文件关闭前，将内容提交到磁盘，而不是在缓存中
          if (this.mutable) { 
              this.flush();
          }
          //关闭文件通道
          this.channel.close();
      }

因为MetaQ的消息都是使用文件存储的，因此具体分析一下FileMessageSet这个类

首先，这个类里面有这些域

public class FileMessageSet implements MessageSet, Closeable { 
    
        private final FileChannel channel;   //对应文件通道
        private final AtomicLong messageCount;  //内容数量
        private final AtomicLong sizeInBytes;
        private final AtomicLong highWaterMark; // 已经确保写入磁盘的水位
        private final long offset; // 镜像offset
        private boolean mutable; // 是否可变
    
        static final Log log = LogFactory.getLog(FileMessageSet.class);

*  说一下文件通道
    
     *  通道既不是一个扩展也不是一项增强，而是全新的、极好的Java I/O示例，提供与I/O服务的直接连接。Channel用于在字节缓冲区和位于通道另一侧的实体（通常是一个文件或套接字）之间有效地传输数据。
     *  一个FileChannel对象却只能通过在一个打开的RandomAccessFile、FileInputStream或FileOutputStream对象上调用getChannel()方法来获取，开发者不能直接创建一个FileChannel。调用getChannel()方法会返回一个连接到相同文件的FileChannel对象且该FileChannel对象具有与File对象相同的访问权限，然后就可以使用通道对象来利用强大的FileChannel API了。
     *  FileChannel对象是线程安全的，多个进程可以在同一个实例上并发调用方法而不会引起任何问题，不过并非所有的操作都是多线程的。影响通道位置或者影响文件的操作都是单线程的，如果有一个线程已经在执行会影响通道位置或文件大小的操作，那么其他尝试进行此类操作之一的线程必须等待，并发行为也会受到底层操作系统或文件系统的影响。
     *  通道只能使用ByteBuffer。write方法写ByteBuffer中的内容至文件中，注意写之前还是要先把ByteBuffer给flip一下。

FileMessageSet构造函数

public FileMessageSet(final FileChannel channel, final long offset, final long limit, final boolean mutable)
                throws IOException { 
            super();
            this.channel = channel;
            this.offset = offset;
            this.messageCount = new AtomicLong(0);
            this.sizeInBytes = new AtomicLong(0);
            this.highWaterMark = new AtomicLong(0);
            this.mutable = mutable;
            if (mutable) { 
                final long startMs = System.currentTimeMillis();
                final long truncated = this.recover();
                if (this.messageCount.get() > 0) { 
                    log.info("Recovery succeeded in " + (System.currentTimeMillis() - startMs) / 1000 + " seconds. "
                            + truncated + " bytes truncated.");
                }
            }
            else { 
                try { 
                    this.sizeInBytes.set(Math.min(channel.size(), limit) - offset);
                    this.highWaterMark.set(this.sizeInBytes.get());
                }
                catch (final Exception e) { 
                    log.error("Set sizeInBytes error", e);
                }
            }
        }

*  首先，传入进来的参数有：文件通道channel、文件偏移offset、文件是否可变mutable
 *  内容数量、磁盘水位都初始化为0，使用AtomicLong能够保证增加的时候是一个原子操作。
 *  如果是可以改变的(mutable为true)，首先记录了一个开始时间，然后执行recover()方法，这个方法主要是验证文件内容的完整性，后面会介绍
 *  当mutable为false的时候，表示文件不能更改，这个时候，sizeInBytes为文件大小(文件通道的偏移—起始偏移)，磁盘水位也设置成sizeInBytes也就是文件大小

FileMessageSet append方法，这个方法是MessageSet接口的append方法的实现。

@Override
        public long append(final ByteBuffer buf) throws IOException { 
            if (!this.mutable) { 
                throw new UnsupportedOperationException("Immutable message set");
            }
            final long offset = this.sizeInBytes.get();
            int sizeInBytes = 0;
            while (buf.hasRemaining()) { 
                sizeInBytes += this.channel.write(buf);
            }
            this.sizeInBytes.addAndGet(sizeInBytes);
            this.messageCount.incrementAndGet();
            return offset;
        }

*  如果不可改变，那么抛出异常
 *  这个时候，并没有将内容写入磁盘，还在通道缓存中，需要在适当的时候，调用flush方法
 *  这个时候，磁盘的水位也是不更新的，只有保证写入磁盘之后，磁盘的水位才更新

FileMessageSet flush方法，刷盘

@Override
        public void flush() throws IOException { 
            this.channel.force(true);
            this.highWaterMark.set(this.sizeInBytes.get());
        }

*  刷盘执行
 *  设置磁盘水位

FileMessageSet slice方法，获取一个消息集合，返回一个MessageSet镜像

@Override
        public MessageSet slice(final long offset, final long limit) throws IOException { 
            return new FileMessageSet(this.channel, offset, limit, false);
        }
    
        static final Log transferLog = LogFactory.getLog("TransferLog");

FileMessageSet read方法

@Override
        public void read(final ByteBuffer bf, final long offset) throws IOException { 
            int size = 0;
            while (bf.hasRemaining()) { 
                final int l = this.channel.read(bf, offset + size);
                if (l < 0) { 
                    break;
                }
                size += l;
            }
        }
    
    
        @Override
        public void read(final ByteBuffer bf) throws IOException { 
            this.read(bf, this.offset);
        }

*  调用了文件通道FileChannel.read方法
 *  首先size为0，后面每次增加读出来的长度，这样能够保证每次读的是最新的
 *  将对应文件通道channel的内容读入到目标字节缓冲区ByteBuffer里面

FileMessageSet write方法

@Override
        public void write(final GetCommand getCommand, final SessionContext ctx) { 
            final IoBuffer buf = this.makeHead(getCommand.getOpaque(), this.sizeInBytes.get());
            // transfer to socket
            this.tryToLogTransferInfo(getCommand, ctx.getConnection());
            ctx.getConnection().transferFrom(buf, null, this.channel, this.offset, this.sizeInBytes.get());
        }
    
    
        public long write(final WritableByteChannel socketChanel) throws IOException { 
            try { 
                return this.getFileChannel().transferTo(this.offset, this.getSizeInBytes(), socketChanel);
            }
            catch (final IOException e) { 
                // Check to see if the IOException is being thrown due to
                // http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=5103988
                final String message = e.getMessage();
                if (message != null && message.contains("temporarily unavailable")) { 
                    return 0;
                }
                throw e;
            }
        }

*  使用了零拷贝将数据从磁盘文件写入socket
 *  零拷贝主要的任务就是避免CPU将数据从一块存储拷贝到另外一块存储，主要就是利用各种零拷贝技术，避免让CPU做大量的数据拷贝任务，减少不必要的拷贝，或者让别的组件(DMA)来做这一类简单的数据传输任务，让CPU解脱出来专注于别的任务。这样就可以让系统资源的利用更加有效。

FileMessageSet recover方法，主要是对内容进行校验，检测文件是否损坏。

private long recover() throws IOException { 
            if (fastBoot) { 
                final long size = this.channel.size();
                this.sizeInBytes.set(size);
                this.highWaterMark.set(size);
                this.messageCount.set(0);
                this.channel.position(size);
                return 0;
            }
            if (!this.mutable) { 
                throw new UnsupportedOperationException("Immutable message set");
            }
            final long len = this.channel.size();
            final ByteBuffer buf = ByteBuffer.allocate(MessageUtils.HEADER_LEN);
            long validUpTo = 0L;
            long next = 0L;
            long msgCount = 0;
            do { 
                next = this.validateMessage(buf, validUpTo, len);
                if (next >= 0) { 
                    msgCount++;
                    validUpTo = next;
                }
            } while (next >= 0);
            this.channel.truncate(validUpTo);
            this.sizeInBytes.set(validUpTo);
            this.highWaterMark.set(validUpTo);
            this.messageCount.set(msgCount);
            this.channel.position(validUpTo);
            return len - validUpTo;
        }

*  fastBoot字段表示是否快速启动：如果这个字段为true，表示不进行后续的校验
 *  validateMessage校验消息md5是否正确
 *  最后再设置磁盘的水位

FileMessageSet validateMessage：校验消息

private long validateMessage(final ByteBuffer buf, final long start, final long len) throws IOException { 
            buf.rewind();
            long read = this.channel.read(buf);
            if (read < MessageUtils.HEADER_LEN) { 
                return -1;
            }
            buf.flip();
            final int messageLen = buf.getInt();
            final long next = start + MessageUtils.HEADER_LEN + messageLen;
            if (next > len) { 
                return -1;
            }
            final int checksum = buf.getInt();
            if (messageLen < 0) { 
                // 数据损坏
                return -1;
            }
    
            final ByteBuffer messageBuffer = ByteBuffer.allocate(messageLen);
            long curr = start + MessageUtils.HEADER_LEN;
            while (messageBuffer.hasRemaining()) { 
                read = this.channel.read(messageBuffer);
                if (read < 0) { 
                    throw new IOException("文件在recover过程中被修改");
                }
                curr += read;
            }
            if (CheckSum.crc32(messageBuffer.array()) != checksum) { 
                return -1;
            }
            else { 
                return next;
            }
        }

*  检验数据是否损坏
 *  在存储之前，将checksum的信息存入磁盘
 *  读取的时候进行校验，比较前后的checksum是否一致，防止篡改

总结：FileMessageSet是MetaQ服务器端存储的一个元素，代表了对一个文件的读写操作，能够对文件内容进行完整性和一致性校验，并能提供统计数据

接下来要介绍的是通过MessageStore怎样的组合将一个个的FileMessageSet

#### 2. MessageStore模块 ####

Broker消息的存储结构：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 5]  
对于每一个topic来说，分区是最小的元素，分区对外的API主要又MessageStore提供，一个MessageStore实例代表着一个分区的实例，分区又存储着具体的内容。

在MetaQ中，分区的存储方式采用多文件的方式进行组合，即MessageStore由多个FileMessageSet组成，而FileMessageSet在MessageStore被包装成Segment，代码如下：

static class Segment { 
            // 该片段代表的offset
            final long start;
            // 对应的文件
            final File file;
            // 该片段的消息集合
            FileMessageSet fileMessageSet;

可以看到有一个fileMessageSet

首先，我们看一下MessageStore的构造函数：

public MessageStore(final String topic, final int partition, final MetaConfig metaConfig,
    			final DeletePolicy deletePolicy, final long offsetIfCreate) throws IOException { 
    	this.metaConfig = metaConfig; //全局配置信息
    	this.topic = topic; //当前主题
    	final TopicConfig topicConfig = this.metaConfig.getTopicConfig(this.topic);
    	String dataPath = metaConfig.getDataPath(); //当前分区的存储路径
    	if (topicConfig != null) { 
    		dataPath = topicConfig.getDataPath();
    	}
    	final File parentDir = new File(dataPath);
    	this.checkDir(parentDir); //检测父目录是否存在
    	this.partitionDir = new File(dataPath + File.separator + topic + "-" + partition);
    	this.checkDir(this.partitionDir);
    	// this.topic = topic;
    	this.partition = partition; //当前分区
    	this.unflushed = new AtomicInteger(0); //未提交的消息数
    	this.lastFlushTime = new AtomicLong(SystemTimer.currentTimeMillis()); //最后一次提交时间
    	this.unflushThreshold = topicConfig.getUnflushThreshold(); //最大允许的未flush消息数，超过此值将强制force到磁盘，默认1000
    	this.deletePolicy = deletePolicy;  //由于是多文件的存储方式，消费过的消息或过期消息需要删除从而腾出空间给新消息的，默认提供归档和过期删除的方式
     
    	// Make a copy to avoid getting it again and again.
    	this.maxTransferSize = metaConfig.getMaxTransferSize();
    	//启动异步写入的时候，消息提交到磁盘的size配置，同时也是配置组写入时，消息最大长度的控制参数，如果消息长度大于该参数，则会同步写入
    	this.maxTransferSize = this.maxTransferSize > ONE_M_BYTES ? ONE_M_BYTES : this.maxTransferSize;
     
    	// Check directory and load exists segments.
    	this.checkDir(this.partitionDir);
    	this.loadSegments(offsetIfCreate);
    	if (this.useGroupCommit()) { 
    		this.start();
    	}
    }

*  首先获取配置信息，有一个全局的配置信息和当前的主题以及当前分区的存储路径
 *  由于MessageStore采用的是多文件的存储方式，因此需要检测父目录是否存在
 *  如果unflushThreshold<0，则启动异步写入的方式
 *  然后检查父目录(checkDir)
 *  最后加载文件(Segments)

加载配置之后，加载了消息文件(loadSegments)，现在我们来分析一下这个方法的源码。

private void loadSegments(final long offsetIfCreate) throws IOException { 
    		final List<Segment> accum = new ArrayList<Segment>();
    		final File[] ls = this.partitionDir.listFiles();
     
    		if (ls != null) { 
                //遍历目录下的所有.meta后缀的数据文件，将所有文件都变为不可变的文件
    			for (final File file : ls) { 
    				if (file.isFile() && file.toString().endsWith(FILE_SUFFIX)) { 
    					if (!file.canRead()) { 
    						throw new IOException("Could not read file " + file);
    					}
    					final String filename = file.getName();
    					final long start = Long.parseLong(filename.substring(0, filename.length() - FILE_SUFFIX.length()));
    					// 先作为不可变的加载进来
    					accum.add(new Segment(start, file, false));
    				}
    			}
    		}
    		
    		if (accum.size() == 0) { 
    			// 没有可用的文件，创建一个，索引从offsetIfCreate开始
    			final File newFile = new File(this.partitionDir, this.nameFromOffset(offsetIfCreate));
    			accum.add(new Segment(offsetIfCreate, newFile));
    		} else { 
    			// 至少有一个文件，校验并按照start升序排序
    			Collections.sort(accum, new Comparator<Segment>() { 
    				@Override
    				public int compare(final Segment o1, final Segment o2) { 
    					if (o1.start == o2.start) { 
    						return 0;
    					} else if (o1.start > o2.start) { 
    						return 1;
    					} else { 
    						return -1;
    					}
    				}
    			});
    			// 校验文件，是否前后衔接，如果不是，则认为数据文件被破坏或者篡改，抛出异常
    			this.validateSegments(accum);
    			// 最后一个文件修改为可变
    			final Segment last = accum.remove(accum.size() - 1);
    			last.fileMessageSet.close();
    			log.info("Loading the last segment in mutable mode and running recover on " + last.file.getAbsolutePath());
    			final Segment mutable = new Segment(last.start, last.file);
    			accum.add(mutable);
    			log.info("Loaded " + accum.size() + " segments...");
    		}
     
    		this.segments = new SegmentList(accum.toArray(new Segment[accum.size()]));
           //多个segmentg通过SegmentList组织起来，SegmentList能保证在并发访问下的删除、添加保持一致性，SegmentList没有采用java的关键字 synchronized进行同步，而是使用类似cvs原语的方式进行同步访问（因为绝大部分情况下并没有并发问题，可以极大的提高效率）。
    }

*  loadSegment方法主要是加载并组织消息文件
 *  首先创建了一个名为accum的List
 *  然后遍历目录下面所有的file.toString().endsWith(FILE\_SUFFIX)也就是.meta结尾的文件，然后将mutable字段设置成为false，也就是不可变文件，加载到accum中
 *  然后对accum这个List中的文件按照start的升序排序，如果没有，那么重新创建
 *  之后再检验文件前后衔接是否正确(validateSegment)
 *  然后将最后一个文件修改为可变文件
 *  可以看到，MessageStore采用Segment方式组织存储，Segment包装了FileMessageSet进行读写，MessageStore并将多个Segment进行前后衔接，衔接方式为：第一个Segment对应的消息文件命名为0.meta，第二个则命名为第一个文件的开始位置+第一个Segment的大小，图示如下(假设现在每个文件大小都为1024byte)，这样设计可以提高查询效率  
    ![在这里插入图片描述][20190707192618808.png]  
    添加消息

//添加消息的方式有两种，同步和异步
    	public void append(final long msgId, final PutCommand req, final AppendCallback cb) { 
    		//首先将内容包装成前面介绍过的消息存储格式
    		this.appendBuffer(MessageUtils.makeMessageBuffer(msgId, req), cb);
    	}
      //异步写入的包装类
    	private static class WriteRequest { 
    		public final ByteBuffer buf;
    		public final AppendCallback cb;
    		public Location result;
     
    		public WriteRequest(final ByteBuffer buf, final AppendCallback cb) { 
    			super();
    			this.buf = buf;
    			this.cb = cb;
    		}
    	}
     
      //这里比较好的设计是采用回调的方式来，对于异步写入实现就变得非常容易
     //AppendCallback返回的是消息成功写入的位置Location(起始位置和消息长度)，该Location并不是相对于当前Segment的开始位置0，而是相对于当前Segment给定的值(对应文件命名值即为给定的值)，以后查询消息的时候直接使用该位置就可以快速定位到消息写入到哪个文件
    //这也就是为什么文件名的命名采用前后衔接的方式，这也通过2分查找可以快速定位消息的位置
    	private void appendBuffer(final ByteBuffer buffer, final AppendCallback cb) { 
    		if (this.closed) { 
    			throw new IllegalStateException("Closed MessageStore.");
    		}
    		//如果启动异步写入并且消息长度小于一次提交的最大值maxTransferSize，则将该消息放入异步写入队列
    		if (this.useGroupCommit() && buffer.remaining() < this.maxTransferSize) { 
    			this.bufferQueue.offer(new WriteRequest(buffer, cb));
    		} else { 
    			Location location = null;
    			final int remainning = buffer.remaining();
    			this.writeLock.lock();
    			try { 
    				final Segment cur = this.segments.last();
    				final long offset = cur.start + cur.fileMessageSet.append(buffer);
    				this.mayBeFlush(1);
    				this.mayBeRoll();
    				location = Location.create(offset, remainning);
    			} catch (final IOException e) { 
    				log.error("Append file failed", e);
    				location = Location.InvalidLocaltion;
    			} finally { 
    				this.writeLock.unlock();
    				if (cb != null) { 
    					cb.appendComplete(location);
    				}
    				//调用回调方法，数据写入文件缓存
    			}
    		}
    	}
     
    ……
     
      //判断是否启用异步写入，如果设置为unflushThreshold <=0的数字，则认为启动异步写入；如果设置为unflushThreshold =1，则是同步写入，即每写入一个消息都会提交到磁盘；如果unflushThreshold>0，则是依赖组提交或者是超时提交
    	private boolean useGroupCommit() { 
    		return this.unflushThreshold <= 0;
    	}
     
    	@Override
    	public void run() { 
    		// 等待force的队列
    		final LinkedList<WriteRequest> toFlush = new LinkedList<WriteRequest>();
    		WriteRequest req = null;
    		long lastFlushPos = 0;
    		Segment last = null;
    		//存储没有关闭并且线程没有被中断
    		while (!this.closed && !Thread.currentThread().isInterrupted()) { 
    			try { 
    				if (last == null) { 
    					//获取最后的一个segment，将消息写入最后segment对应的文件
    					last = this.segments.last();
    					lastFlushPos = last.fileMessageSet.highWaterMark();
    				}
    				if (req == null) { 
                         //如果等待提交到磁盘的队列toFlush为空，则两种可能：一、刚刚提交完，列表为空；二、等待写入消息的队列为空，如果判断toFlush，则调用bufferQueue.take()方法，可以阻塞住队列，而如果toFlush不为空，则调用bufferQueue.poll，这是提高性能的一种做法。
    					if (toFlush.isEmpty()) { 
    						req = this.bufferQueue.take();
    					} else { 
    						req = this.bufferQueue.poll();
    						//如果当前请求为空，表明等待写入的消息已经没有了，这时候文件缓存中的消息需要提交到磁盘，防止消息丢失；或者如果已经写入文件的大小大于maxTransferSize，则提交到磁盘
                              //这里需要注意的是，会出现这样一种情况，刚好最后一个segment的文件快满了，这时候是不会roll出一个新的segment写入消息的，而是直接追加到原来的segment尾部，可能导致segment对应的文件大小大于配置的单个segment大小
    						if (req == null || last.fileMessageSet.getSizeInBytes() > lastFlushPos + this.maxTransferSize) { 
    							// 强制force，确保内容保存到磁盘
    							last.fileMessageSet.flush();
    							lastFlushPos = last.fileMessageSet.highWaterMark();
    							// 通知回调
    							//异步写入比组写入可靠，因为异步写入一定是提交到磁盘的时候才进行回调的，而组写入如果依赖组提交的方式，则可能会丢失数据，因为组写入在消息写入到文件缓存的时候就进行回调了(除非设置unflushThreshold=1)
    							for (final WriteRequest request : toFlush) { 
    								request.cb.appendComplete(request.result);
    							}
    							toFlush.clear();
    							// 是否需要roll
    							this.mayBeRoll();
    							// 如果切换文件，重新获取last
    							if (this.segments.last() != last) { 
    								last = null;
    							}
    							continue;
    						}
    					}
    				}
     
    				if (req == null) { 
    					continue;
    				}
    				//写入文件，并计算写入位置
    				final int remainning = req.buf.remaining();
    				//写入位置为：当前segment给定的值 + 加上文件已有的长度 
    final long offset = last.start + last.fileMessageSet.append(req.buf);
    				req.result = Location.create(offset, remainning);
    				if (req.cb != null) { 
    					toFlush.add(req);
    				}
    				req = null;
    			} catch (final IOException e) { 
    				log.error("Append message failed,*critical error*,the group commit thread would be terminated.", e);
    				// TODO io异常没办法处理了，简单跳出?
    				break;
    			} catch (final InterruptedException e) { 
    				// ignore
    			}
    		}
    		// terminated
    		//关闭store 前，将等待写入队列中的剩余消息写入最后一个文件，这时候如果最后一个Segment满了也不会roll出新的Segment，持续的将消息写入到最后一个Segment，所以这时候也会发生Segment的size大于配置的size的情况
    		try { 
    			for (WriteRequest request : this.bufferQueue) { 
    				final int remainning = request.buf.remaining();
    				final long offset = last.start + last.fileMessageSet.append(request.buf);
    				if (request.cb != null) { 
    					request.cb.appendComplete(Location.create(offset, remainning));
    				}
    			}
    			this.bufferQueue.clear();
    		} catch (IOException e) { 
    			log.error("Append message failed", e);
    		}
    	}
    ……
     
    	 //Append多个消息，返回写入的位置
    	public void append(final List<Long> msgIds, final List<PutCommand> putCmds, final AppendCallback cb) { 
    		this.appendBuffer(MessageUtils.makeMessageBuffer(msgIds, putCmds), cb);
    	}
     
    	/** * 重放事务操作，如果消息没有存储成功，则重新存储，并返回新的位置 */
    	public void replayAppend(final long offset, final int length, final int checksum, final List<Long> msgIds,
    			final List<PutCommand> reqs, final AppendCallback cb) throws IOException { 
           //如果消息没有存储，则重新存储，写到最后一个Segment尾部
    		final Segment segment = this.findSegment(this.segments.view(), offset);
    		if (segment == null) { 
    			this.append(msgIds, reqs, cb);
    		} else { 
    			final MessageSet messageSet = segment.fileMessageSet.slice(offset - segment.start, offset - segment.start + length);
    			final ByteBuffer buf = ByteBuffer.allocate(length);
    			messageSet.read(buf, offset - segment.start);
    			buf.flip();
    			final byte[] bytes = new byte[buf.remaining()];
    			buf.get(bytes);
    			// 这个校验和是整个消息的校验和，这跟message的校验和不一样，注意区分
    			final int checkSumInDisk = CheckSum.crc32(bytes);
    			// 没有存入，则重新存储
    			if (checksum != checkSumInDisk) { 
    				this.append(msgIds, reqs, cb);
    			} else { 
    				// 正常存储了消息，无需处理
    				if (cb != null) {  
    					this.notifyCallback(cb, null);
    				}
    			}
    		}
    	}
     
    //判断是否需要roll，如果当前 messagestore最后一个segment的size>=配置的segment size，则产生新的segment，并将新的segment作为最后一个segment，原来最后的segment提交一次，并将mutable设置为false
    private void mayBeRoll() throws IOException { 
    		if (this.segments.last().fileMessageSet.getSizeInBytes() >= this.metaConfig.getMaxSegmentSize()) { 
    			this.roll();
    		}
    	}
     
    	String nameFromOffset(final long offset) { 
    		final NumberFormat nf = NumberFormat.getInstance();
    		nf.setMinimumIntegerDigits(20);
    		nf.setMaximumFractionDigits(0);
    		nf.setGroupingUsed(false);
    		return nf.format(offset) + FILE_SUFFIX;
    	}
     
    private long nextAppendOffset() throws IOException { 
    		final Segment last = this.segments.last();
    		last.fileMessageSet.flush();
    		return last.start + last.size();
    	}
     
    	private void roll() throws IOException { 
    		final long newOffset = this.nextAppendOffset();
    		//新的segment对应的存储文件的命名为原来最后一个segment的起始位置 + segment的size
    		final File newFile = new File(this.partitionDir, this.nameFromOffset(newOffset));
    		this.segments.last().fileMessageSet.flush();
    		this.segments.last().fileMessageSet.setMutable(false);
    		this.segments.append(new Segment(newOffset, newFile));
    	}
     
    //判断是否需要消息提交到磁盘，判断的条件有两个，如果达到组提交的条件或者达到间隔的提交时间
    private void mayBeFlush(final int numOfMessages) throws IOException { 
    		if (this.unflushed.addAndGet(numOfMessages) > this.metaConfig.getTopicConfig(this.topic).getUnflushThreshold()
    				|| SystemTimer.currentTimeMillis() - this.lastFlushTime.get() > this.metaConfig.getTopicConfig(this.topic).getUnflushInterval()) { 
    			this.flush0();
    		}
    	}
     
    //提交到磁盘
    public void flush() throws IOException { 
    		this.writeLock.lock();
    		try { 
    			this.flush0();
    		} finally { 
    			this.writeLock.unlock();
    		}
    	}
     
    	
    private void flush0() throws IOException { 
    		if (this.useGroupCommit()) { 
    			return;
    		}
    		//由于只有最后一个segment是可变，即可写入消息的，所以只需要提交最后一个segment的消息
    		this.segments.last().fileMessageSet.flush();
    		this.unflushed.set(0);
    		this.lastFlushTime.set(SystemTimer.currentTimeMillis());
    	}
     
     
    @Override
    	public void close() throws IOException { 
    		this.closed = true;
    		this.interrupt();
         //等待子线程完成写完异步队列中剩余未写的消息
    		try { 
    			this.join(500);
    		} catch (InterruptedException e) { 
    			Thread.currentThread().interrupt();
    		}
          //关闭segment，保证内容都已经提交到磁盘
    		for (final Segment segment : this.segments.view()) { 
    			segment.fileMessageSet.close();
    		}
    	}
     
    //返回segment的信息，主要包括segment的开始位置以及 segment 的size
    public List<SegmentInfo> getSegmentInfos() { 
    		final List<SegmentInfo> rt = new ArrayList<SegmentInfo>();
    		for (final Segment seg : this.segments.view()) { 
    			rt.add(new SegmentInfo(seg.start, seg.size()));
    		}
    		return rt;
    	}
     
    /** * 返回当前最大可读的offset */
      //需要注意的是，在文件缓存中的消息是不可读的，可以通过getSizeInBytes（）方法来判断还有多少内容还在文件缓存中，getSizeInBytes()方法返回的值是包括所有在磁盘和缓存中的size
    	public long getMaxOffset() { 
    		final Segment last = this.segments.last();
    		if (last != null) { 
    			return last.start + last.size();
    		} else { 
    			return 0;
    		}
    	}
     
    	/** * 返回当前最小可读的offset */
    	public long getMinOffset() { 
    		Segment first = this.segments.first();
    		if (first != null) { 
    			return first.start;
    		} else { 
    			return 0;
    		}
    	}
     
    /** * 根据offset和maxSize返回所在MessageSet, 当offset超过最大offset的时候返回null， * 当offset小于最小offset的时候抛出ArrayIndexOutOfBounds异常 */
      //代码的注释以及清楚的解析了作用
    	public MessageSet slice(final long offset, final int maxSize) throws IOException { 
    		final Segment segment = this.findSegment(this.segments.view(), offset);
    		if (segment == null) { 
    			return null;
    		} else { 
    			return segment.fileMessageSet.slice(offset - segment.start, offset - segment.start + maxSize);
    		}
    	}
     
    /** * 根据offset查找文件,如果超过尾部，则返回null，如果在头部之前，则抛出ArrayIndexOutOfBoundsException */
      //指定位置找到对应的segment，由于前面的文件组织方式，所以这里可以采用2分查找的方式，
      //效率很高
    	Segment findSegment(final Segment[] segments, final long offset) { 
    		if (segments == null || segments.length < 1) { 
    			return null;
    		}
    		// 老的数据不存在，返回最近最老的数据
    		final Segment last = segments[segments.length - 1];
    		// 在头部以前，抛出异常
    		if (offset < segments[0].start) { 
    			throw new ArrayIndexOutOfBoundsException();
    		}
    		// 刚好在尾部或者超出范围，返回null
    		if (offset >= last.start + last.size()) { 
    			return null;
    		}
    		// 根据offset二分查找
    		int low = 0;
    		int high = segments.length - 1;
    		while (low <= high) { 
    			final int mid = high + low >>> 1;
    			final Segment found = segments[mid];
    			if (found.contains(offset)) { 
    				return found;
    			} else if (offset < found.start) { 
    				high = mid - 1;
    			} else { 
    				low = mid + 1;
    			}
    		}
    		return null;
    	}

这一段代码需要重点看

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70]: /images/20211219/c8ab7bc6b5ce45da836eaf45ed27a1d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 1]: /images/20211219/cbd72ecb645b4901adf0cbb55a03e7cd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 2]: /images/20211219/938e2d1ff8254725a0a2c644282077a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 3]: /images/20211219/773b7e8b29b346929df235832b9a6f12.png
[20190707123445343.png]: /images/20211219/bf8cbb2da87b4c06b7d86a90b4d05e89.png
[20190707123943638.png]: /images/20211219/bdc34469b969453996b67b82ca2f18db.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 4]: /images/20211219/2e4417bd5e4a45eeb977b1e26ad1b90f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 5]: /images/20211219/dbbaa08d0ed740139943283f384b0e06.png
[20190707192618808.png]: /images/20211219/34d031ba36d84e8dba454f6fe8a96fb2.png