MetaQ技术内幕——源码分析(转)

笔者最近在业务上需要使用到MetaQ，也借此阅读了MetaQ的相关源码，准备分享MetaQ源码分析。先扫扫盲，如果读者对MetaQ已经较为熟悉，可以跳过下一段落。

 

一、MetaQ简介

 

MetaQ（全称Metamorphosis）是一个高性能、高可用、可扩展的分布式消息中间件，,MetaQ具有消息存储顺序写、吞吐量大和支持本地和XA事务等特性，适用于大吞吐量、顺序消息、广播和日志数据传输等场景，METAQ在阿里巴巴各个子公司被广泛应用，每天转发250亿+条消息。主要应用于异步解耦，Mysql数据复制，收集日志等场景。

 

 主要特点

 

• 生产者、服务器和消费者都可分布
• 消息存储顺序写
• 性能极高,吞吐量大
• 支持消息顺序
• 支持本地和XA事务
• 客户端pull，随机读,利用sendfile系统调用，zero-copy ,批量拉数据
• 支持消费端事务
• 支持消息广播模式
• 支持异步发送消息
• 支持http协议
• 支持消息重试和recover
• 数据迁移、扩容对用户透明
• 消费状态保存在客户端
• 支持同步和异步复制两种HA

二、MetaQ的总体架构

 

消息中间件消费消息有两种方式，推送消息和拉取消息，一般对于实时性要求非常高的消息中间件，会更多的采用推送的模式，如ActivitiQ，而MetaQ采用的主动拉取消息的模式，通过Zookeeper进行协调，MetaQ分为三部分：Server、Client（包括消息产生者和消费者）以及协调者Zookeeper，总体结构如下：

 

 

 

三、工程结构分析

 

MetaQ的源码可以从以下链接获取：https://github.com/killme2008/Metamorphosis （本序列基于MetaQ 1.4.5进行分析），下载工程到本地，我们发现主要有如下几个工程，先大致介绍一下工程的划分：

 

 

• 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，提供服务端管理和操作的一些工具

 工程依赖图大致如下：

 

 

 

源码分析也将以分Server和Client两阶段进行，将优先进行Server的分析，下一篇将正式带大家进入MetaQ的世界。

消息，是MetaQ最重要的资源，在分析MetaQ之前必须了解的概念，我们所做的一切都是围绕消息进行的，让我们看看MetaQ中消息的定义是怎样的，MetaQ的类Message定义了消息的格式：

Java代码  

1. public class Message implements Serializable {  
2.   private long id; //消息的ID  
3.   private String topic; //消息属于哪个主题  
4.   private byte[] data;  //消息的内容  
5.   private String attribute; //消息的属性  
6.   private int flag; //属性标志位，如果属性不为空，则该标志位true  
7.   private Partition partition; //该主题下的哪个分区，简单的理解为发送到该主题下的哪个队列  
8.   private transient boolean readOnly; //消息是否只读  
9.   private transient boolean rollbackOnly = false; //该消息是否需要回滚，主要用于事务实现的需要  
10. }  

从对消息的定义，我们可以看出消息都有一个唯一的ID，并且归属于某个主题，发送到该主题下的某个分区，具有一些基本属性。

 

MetaQ分为Broker和Client以及Zookeeper，系统结构如下：

 

 

 

 

下面我们先分析Broker源码。

 

Broker主要围绕发送消息和消费消息的主线进行，对于Broker来说就是输入、输出流的处理。在该主线下，Broker主要分为如下模块：网络传输模块、消息存储模块、消息统计模块以及事务模块，本篇首先针对独立性较强的消息存储模块进行分析。

 

在进行存储模块分析之前，我们得了解Broker中的一个重要的类MetaConfig，MetaConfig是Broker配置加载器，通过MetaConfig可以获取到各模块相关的配置，所以MetaConfig是贯穿所有模块的类。MetaConfig实现MetaConfigMBean接口，该接口定义如下：

Java代码  

1. public interface MetaConfigMBean {  
2.     /** 
3.      * Reload topics configuration 
4.      */  
5.     public void reload();  
6.    
7.     /**关闭分区 */  
8.     public void closePartitions(String topic, int start, int end);  
9.    
10.     /**打开一个topic的所有分区 */  
11.     public void openPartitions(String topic);  
12. }  

 

MetaConfig注册到了MBeanServer上，所以可以通过JMX协议重新加载配置以及关闭和打开分区。为了加载的配置立即生效，MetaConfig内置了一个通知机制，可以通过向MetaConfig注册监听器的方式关注相关配置的变化，监听器需实现PropertyChangeListener接口。

Java代码  

1. public void addPropertyChangeListener(final String propertyName, final PropertyChangeListener listener) {  
2.       this.propertyChangeSupport.addPropertyChangeListener(propertyName, listener);  
3. }  
4.    
5. public void removePropertyChangeListener(final PropertyChangeListener listener) {  
6.       this.propertyChangeSupport.removePropertyChangeListener(listener);  
7. }  

 

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

Java代码  

1. //configFileChecksum通知  
2. private Ini createIni(final File file) throws IOException, InvalidFileFormatException {  
3.       ……  
4.       this.propertyChangeSupport.firePropertyChange("configFileChecksum", null, null);  
5.       return conf;  
6. }  
7.    
8. public void setConfigFileChecksum(long configFileChecksum) {  
9.       this.configFileChecksum = configFileChecksum;  
10.       this.propertyChangeSupport.firePropertyChange("configFileChecksum", null, null);  
11. }  
12.    
13. //topics、topicConfigMap和unflushInterval通知  
14. private void populateTopicsConfig(final Ini conf) {  
15. ……  
16. if (!newTopicConfigMap.equals(this.topicConfigMap)) {  
17.           this.topics = newTopics;  
18.           this.topicConfigMap = newTopicConfigMap;  
19.           this.propertyChangeSupport.firePropertyChange("topics", null, null);  
20.           this.propertyChangeSupport.firePropertyChange("topicConfigMap", null, null);  
21.   }  
22.   this.propertyChangeSupport.firePropertyChange("unflushInterval", null, null);  
23. ……  

需要注意的是，调用reload方法时，只对topic的配置生效，对全局配置不生效，只重载topic的配置。

 

好吧，废话了许多，让我们正式进入存储模块的分析吧。

 

Broker的存储模块用于存储Client发送的等待被消费的消息，Broker采用文件存储的方式来存储消息，存储模块类图如下：

 

  

 

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

 

Java代码  

1. /** 
2.  * 消息集合 
3.  */  
4. public interface MessageSet {  
5.   public MessageSet slice(long offset, long limit) throws IOException; //获取一个消息集合  
6.    
7.   public void write(GetCommand getCommand, SessionContext ctx);  
8.    
9.   public long append(ByteBuffer buff) throws IOException; //存储一个消息，这时候还没有存储到磁盘，需要调用flush方法才能保证存储到磁盘  
10.    
11.   public void flush() throws IOException; //提交到磁盘  
12.    
13.   public void read(final ByteBuffer bf, long offset) throws IOException; //读取消息  
14.    
15.   public void read(final ByteBuffer bf) throws IOException; //读取消息  
16.    
17.   public long getMessageCount();//该集合的消息数量  
18. }  

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

 

Java代码  

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

下面让我们来具体分析一下FileMessageSet这个类，

Java代码  

1. public class FileMessageSet implements MessageSet, Closeable {  
2.     ……  
3.   private final FileChannel channel; //对应的文件通道  
4.   private final AtomicLong messageCount; //内容数量  
5.   private final AtomicLong sizeInBytes;   
6.   private final AtomicLong highWaterMark; // 已经确保写入磁盘的水位  
7.   private final long offset; // 镜像offset  
8.   private boolean mutable;   // 是否可变  
9.    
10.   public FileMessageSet(final FileChannel channel, final long offset, final long limit, final boolean mutable) throws IOException {  
11.       this.channel = channel;  
12.       this.offset = offset;  
13.       this.messageCount = new AtomicLong(0);  
14.       this.sizeInBytes = new AtomicLong(0);  
15.       this.highWaterMark = new AtomicLong(0);  
16.       this.mutable = mutable;  
17.       if (mutable) {   
18.           final long startMs = System.currentTimeMillis();  
19.           final long truncated = this.recover();  
20.           if (this.messageCount.get() > 0) {  
21.               log.info("Recovery succeeded in " + (System.currentTimeMillis() - startMs) / 1000 + " seconds. " + truncated + " bytes truncated.");  
22.           }  
23.       } else {  
24.           try {  
25.               this.sizeInBytes.set(Math.min(channel.size(), limit) - offset);  
26.               this.highWaterMark.set(this.sizeInBytes.get());  
27.           } catch (final Exception e) {  
28.               log.error("Set sizeInBytes error", e);  
29.           }  
30.       }  
31.   }  
32. //注意FileMessageSet的mutable属性，如果mutable为true的时候，将会调用recover()方法，该方法主要是验证文件内容的完整性，后面会详细介绍，如果mutable为false的时候，表明该文件不可更改，这个时候磁盘水位和sizeInBytes的值均为文件大小。  
33.    
34. ……  
35. public long append(final ByteBuffer buf) throws IOException {  
36.       //如果mutable属性为false的时候，不允许追加消息在文件尾  
37.       if (!this.mutable) {  
38.           throw new UnsupportedOperationException("Immutable message set");  
39.       }  
40.       final long offset = this.sizeInBytes.get();  
41.       int sizeInBytes = 0;  
42.       while (buf.hasRemaining()) {  
43.           sizeInBytes += this.channel.write(buf);  
44.       }  
45.       //在这个时候并没有将内容写入磁盘，还在通道的缓存中，需要在适当的时候调用flush方法  
46.        //这个时候磁盘的水位是不更新的，只有在保证写入磁盘才会更新磁盘的水位信息  
47.       this.sizeInBytes.addAndGet(sizeInBytes); 、  
48.       this.messageCount.incrementAndGet();  
49.       return offset;  
50.   }  
51.    
52.    //提交到磁盘  
53.   public void flush() throws IOException {  
54.       this.channel.force(true);  
55.       this.highWaterMark.set(this.sizeInBytes.get());  
56.   }  
57. ……  
58. @Override  
59.   public MessageSet slice(final long offset, final long limit) throws IOException {  
60.       //返回消息集  
61. return new FileMessageSet(this.channel, offset, limit, false);  
62.   }  
63.    
64.   static final Logger transferLog = LoggerFactory.getLogger("TransferLog");  
65.    
66.   @Override  
67.   public void read(final ByteBuffer bf, final long offset) throws IOException {  
68.       //读取内容  
69. int size = 0;  
70.       while (bf.hasRemaining()) {  
71.           final int l = this.channel.read(bf, offset + size);  
72.           if (l < 0) {  
73.               break;  
74.           }  
75.           size += l;  
76.       }  
77.   }  
78.    
79.   @Override  
80.   public void read(final ByteBuffer bf) throws IOException {  
81.       this.read(bf, this.offset);  
82.   }  
83.    
84.  //下面的write方法主要是提供zero拷贝  
85.   @Override  
86.   public void write(final GetCommand getCommand, final SessionContext ctx) {  
87.       final IoBuffer buf = this.makeHead(getCommand.getOpaque(), this.sizeInBytes.get());  
88.       // transfer to socket  
89.       this.tryToLogTransferInfo(getCommand, ctx.getConnection());  
90.       ctx.getConnection().transferFrom(buf, null, this.channel, this.offset, this.sizeInBytes.get());  
91.   }  
92.    
93.   public long write(final WritableByteChannel socketChanel) throws IOException {  
94.       try {  
95.           return this.getFileChannel().transferTo(this.offset, this.getSizeInBytes(), socketChanel);  
96.       } catch (final IOException e) {  
97.           // Check to see if the IOException is being thrown due to  
98.           // http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=5103988  
99.           final String message = e.getMessage();  
100.           if (message != null && message.contains("temporarily unavailable")) {  
101.               return 0;  
102.           }  
103.           throw e;  
104.       }  
105.   }  
106. ……  
107. private static boolean fastBoot = Boolean.valueOf(System.getProperty("meta.fast_boot", "false"));  
108.    
109.   private long recover() throws IOException {  
110.        //如果在System属性里设置了 meta.fast_boot为true，表示快速启动，快速启动不检测文件是否损坏，不对内容进行校验  
111.       if (fastBoot) {  
112.           final long size = this.channel.size();  
113.           this.sizeInBytes.set(size);  
114.           this.highWaterMark.set(size);  
115.           this.messageCount.set(0);  
116.           this.channel.position(size);  
117.           return 0;  
118.       }  
119.       if (!this.mutable) {  
120.           throw new UnsupportedOperationException("Immutable message set");  
121.       }  
122.       final long len = this.channel.size();  
123.       final ByteBuffer buf = ByteBuffer.allocate(MessageUtils.HEADER_LEN);  
124.       long validUpTo = 0L;  
125.       long next = 0L;  
126.       long msgCount = 0;  
127.       do {  
128.           next = this.validateMessage(buf, validUpTo, len);  
129.           if (next >= 0) {  
130.               msgCount++;  
131.               validUpTo = next;  
132.           }  
133.       } while (next >= 0);  
134.       this.channel.truncate(validUpTo);  
135.       this.sizeInBytes.set(validUpTo);  
136.       this.highWaterMark.set(validUpTo);  
137.       this.messageCount.set(msgCount);  
138.       this.channel.position(validUpTo);  
139.       return len - validUpTo;  
140.   }  
141.    
142. 消息在磁盘上存储结构如下：  
143.   /* 
144.    * 20个字节的头部 * 
145.    * <ul> 
146.    * <li>message length(4 bytes),including attribute and payload</li> 
147.    * <li>checksum(4 bytes)</li> 
148.    * <li>message id(8 bytes)</li> 
149.    * <li>message flag(4 bytes)</li> 
150.    * </ul> 
151.    *  length长度的内容 
152.       */  
153.   //在存储之前将checksum的信息存入到磁盘，读取的时候再进行校验，比较前后的  
154.   //checksum是否一致，防止消息被篡改  
155.   private long validateMessage(final ByteBuffer buf, final long start, final long len) throws IOException {  
156.       buf.rewind();  
157.       long read = this.channel.read(buf);  
158.       if (read < MessageUtils.HEADER_LEN) {  
159.           return -1;  
160.       }  
161.       buf.flip();  
162.       final int messageLen = buf.getInt();  
163.       final long next = start + MessageUtils.HEADER_LEN + messageLen;  
164.       if (next > len) {  
165.           return -1;  
166.       }  
167.       final int checksum = buf.getInt();  
168.       if (messageLen < 0) {  
169.           // 数据损坏  
170.           return -1;  
171.       }  
172.    
173.       final ByteBuffer messageBuffer = ByteBuffer.allocate(messageLen);  
174. //        long curr = start + MessageUtils.HEADER_LEN;  
175.       while (messageBuffer.hasRemaining()) {  
176.           read = this.channel.read(messageBuffer);  
177.           if (read < 0) {  
178.               throw new IOException("文件在recover过程中被修改");  
179.           }  
180. //            curr += read;  
181.       }  
182.       if (CheckSum.crc32(messageBuffer.array()) != checksum) {  
183.           //采用crc32对内容进行校验是否一致  
184.           return -1;  
185.       } else {  
186.           return next;  
187.       }  
188.   }  

    FileMessageSet是MetaQ 服务器端存储的一个元素，代表了对一个文件的读写操作，能够对文件内容进行完整性和一致性校验，并能提供统计数据，接下来要介绍的是通过MessageStore怎样的组合将一个个的FileMessageSet，单纯的讲，MetaQ的存储非常值得借鉴。

前面忘了先介绍一下Broker消息存储的组织方式，我们前面知道了一条消息属于某个Topic下的某个分区，消息存储的组织方式是按照此方式进行组织的，结构图如下：

 

 

所以对于每个Topic而言，分区是最小的元素，对外API主要由MessageStore提供，一个MessageStore实例代表一个分区的实例，分区存储具体的内容。在MetaQ中，分区的存储采用的多文件的方式进行组合，即MessageStore由多个FileMessageSet组成，而FileMessageSet在MessageStore被包装成Segment，代码如下（MessageStore是值得好好分析的一个类）：

 

 

Java代码  

1. public class MessageStore extends Thread implements Closeable {  
2.      …….  
3. }  

MessageStore继承了Thread类，继承该类主要是为了实现异步写入方式 

 

Java代码  

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

首先是获取配置信息，其次由于MessageStore采用的多文件存储方,所以要检查父目录的存在，最后则是加载校验已有数据，如果配置了异步写入，则启动异步写入线程（如果unflushThreshold<= 0，则认为启动异步写入的方式）

 

我们发现在构造方法的倒数第3行调用了loadSegments()方法去加载校验文件，看看该方法到底做了些什么事情

 

Java代码  

1. private void loadSegments(final long offsetIfCreate) throws IOException {  
2.         final List<Segment> accum = new ArrayList<Segment>();  
3.         final File[] ls = this.partitionDir.listFiles();  
4.   
5.         if (ls != null) {  
6.             //遍历目录下的所有.meta后缀的数据文件，将所有文件都变为不可变的文件  
7.             for (final File file : ls) {  
8.                 if (file.isFile() && file.toString().endsWith(FILE_SUFFIX)) {  
9.                     if (!file.canRead()) {  
10.                         throw new IOException("Could not read file " + file);  
11.                     }  
12.                     final String filename = file.getName();  
13.                     final long start = Long.parseLong(filename.substring(0, filename.length() - FILE_SUFFIX.length()));  
14.                     // 先作为不可变的加载进来  
15.                     accum.add(new Segment(start, file, false));  
16.                 }  
17.             }  
18.         }  
19.           
20.         if (accum.size() == 0) {  
21.             // 没有可用的文件，创建一个，索引从offsetIfCreate开始  
22.             final File newFile = new File(this.partitionDir, this.nameFromOffset(offsetIfCreate));  
23.             accum.add(new Segment(offsetIfCreate, newFile));  
24.         } else {  
25.             // 至少有一个文件，校验并按照start升序排序  
26.             Collections.sort(accum, new Comparator<Segment>() {  
27.                 @Override  
28.                 public int compare(final Segment o1, final Segment o2) {  
29.                     if (o1.start == o2.start) {  
30.                         return 0;  
31.                     } else if (o1.start > o2.start) {  
32.                         return 1;  
33.                     } else {  
34.                         return -1;  
35.                     }  
36.                 }  
37.             });  
38.             // 校验文件，是否前后衔接，如果不是，则认为数据文件被破坏或者篡改，抛出异常  
39.             this.validateSegments(accum);  
40.             // 最后一个文件修改为可变  
41.             final Segment last = accum.remove(accum.size() - 1);  
42.             last.fileMessageSet.close();  
43.             log.info("Loading the last segment in mutable mode and running recover on " + last.file.getAbsolutePath());  
44.             final Segment mutable = new Segment(last.start, last.file);  
45.             accum.add(mutable);  
46.             log.info("Loaded " + accum.size() + " segments...");  
47.         }  
48.   
49.         this.segments = new SegmentList(accum.toArray(new Segment[accum.size()]));  
50.        //多个segmentg通过SegmentList组织起来，SegmentList能保证在并发访问下的删除、添加保持一致性，SegmentList没有采用java的关键字 synchronized进行同步，而是使用类似cvs原语的方式进行同步访问（因为绝大部分情况下并没有并发问题，可以极大的提高效率），该类比较简单就不再分析  
51. }  

 

MessageStore采用Segment方式组织存储，Segment包装了FileMessageSet，由FileMessageSet进行读写，MessageStore并将多个Segment进行前后衔接，衔接方式为：第一个Segment对应的消息文件命名为0.meta，第二个则命名为第一个文件的开始位置+第一个Segment的大小，图示如下(假设现在每个文件大小都为1024byte)：

 

为什么要这样进行设计呢，主要是为了提高查询效率。MessageStore将最后一个Segment变为可变Segment，因为最后一个Segment相当于文件尾，消息是有先后顺序的，必须将消息添加到最后一个Segment上。

 关注validateSegments()方法做了些什么事情

 

Java代码  

1. private void validateSegments(final List<Segment> segments) {  
2.         //验证按升序排序的Segment是否前后衔接，确保文件没有被篡改和破坏(这里的验证是比较简单的验证，消息内容的验证在FileMessageSet中，通过比较checksum进行验证，在前面的篇幅中介绍过，这两种方式结合可以在范围上从大到小进行验证，保证内容基本不被破坏和篡改)  
3.         this.writeLock.lock();  
4.         try {  
5.             for (int i = 0; i < segments.size() - 1; i++) {  
6.                 final Segment curr = segments.get(i);  
7.                 final Segment next = segments.get(i + 1);  
8.                 if (curr.start + curr.size() != next.start) {  
9.                     throw new IllegalStateException("The following segments don't validate: "  
10.                             + curr.file.getAbsolutePath() + ", " + next.file.getAbsolutePath());  
11.                 }  
12.             }  
13.         } finally {  
14.             this.writeLock.unlock();  
15.         }  
16. }  

 

 

   ITEye整理格式好麻烦，下面的代码分析直接在代码中分析

 

 

Java代码  

1. //添加消息的方式有两种，同步和异步  
2.     public void append(final long msgId, final PutCommand req, final AppendCallback cb) {  
3.         //首先将内容包装成前面介绍过的消息存储格式  
4.         this.appendBuffer(MessageUtils.makeMessageBuffer(msgId, req), cb);  
5.     }  
6.   //异步写入的包装类  
7.     private static class WriteRequest {  
8.         public final ByteBuffer buf;  
9.         public final AppendCallback cb;  
10.         public Location result;  
11.   
12.         public WriteRequest(final ByteBuffer buf, final AppendCallback cb) {  
13.             super();  
14.             this.buf = buf;  
15.             this.cb = cb;  
16.         }  
17.     }  
18.   
19.   //这里比较好的设计是采用回调的方式来，对于异步写入实现就变得非常容易  
20.  //AppendCallback返回的是消息成功写入的位置Location(起始位置和消息长度)，该Location并不是相对于当前Segment的开始位置0，而是相对于当前Segment给定的值(对应文件命名值即为给定的值)，以后查询消息的时候直接使用该位置就可以快速定位到消息写入到哪个文件  
21. //这也就是为什么文件名的命名采用前后衔接的方式，这也通过2分查找可以快速定位消息的位置  
22.     private void appendBuffer(final ByteBuffer buffer, final AppendCallback cb) {  
23.         if (this.closed) {  
24.             throw new IllegalStateException("Closed MessageStore.");  
25.         }  
26.         //如果启动异步写入并且消息长度小于一次提交的最大值maxTransferSize，则将该消息放入异步写入队列  
27.         if (this.useGroupCommit() && buffer.remaining() < this.maxTransferSize) {  
28.             this.bufferQueue.offer(new WriteRequest(buffer, cb));  
29.         } else {  
30.             Location location = null;  
31.             final int remainning = buffer.remaining();  
32.             this.writeLock.lock();  
33.             try {  
34.                 final Segment cur = this.segments.last();  
35.                 final long offset = cur.start + cur.fileMessageSet.append(buffer);  
36.                 this.mayBeFlush(1);  
37.                 this.mayBeRoll();  
38.                 location = Location.create(offset, remainning);  
39.             } catch (final IOException e) {  
40.                 log.error("Append file failed", e);  
41.                 location = Location.InvalidLocaltion;  
42.             } finally {  
43.                 this.writeLock.unlock();  
44.                 if (cb != null) {  
45.                     cb.appendComplete(location);  
46.                 }  
47.                 //调用回调方法，数据写入文件缓存  
48.             }  
49.         }  
50.     }  
51.   
52. ……  
53.   
54.   //判断是否启用异步写入，如果设置为unflushThreshold <=0的数字，则认为启动异步写入；如果设置为unflushThreshold =1，则是同步写入，即每写入一个消息都会提交到磁盘；如果unflushThreshold>0，则是依赖组提交或者是超时提交  
55.     private boolean useGroupCommit() {  
56.         return this.unflushThreshold <= 0;  
57.     }  
58.   
59.     @Override  
60.     public void run() {  
61.         // 等待force的队列  
62.         final LinkedList<WriteRequest> toFlush = new LinkedList<WriteRequest>();  
63.         WriteRequest req = null;  
64.         long lastFlushPos = 0;  
65.         Segment last = null;  
66.         //存储没有关闭并且线程没有被中断  
67.         while (!this.closed && !Thread.currentThread().isInterrupted()) {  
68.             try {  
69.                 if (last == null) {  
70.                     //获取最后的一个segment，将消息写入最后segment对应的文件  
71.                     last = this.segments.last();  
72.                     lastFlushPos = last.fileMessageSet.highWaterMark();  
73.                 }  
74.                 if (req == null) {  
75.                      //如果等待提交到磁盘的队列toFlush为空，则两种可能：一、刚刚提交完，列表为空；二、等待写入消息的队列为空，如果判断toFlush，则调用bufferQueue.take()方法，可以阻塞住队列，而如果toFlush不为空，则调用bufferQueue.poll，这是提高性能的一种做法。  
76.                     if (toFlush.isEmpty()) {  
77.                         req = this.bufferQueue.take();  
78.                     } else {  
79.                         req = this.bufferQueue.poll();  
80.                         //如果当前请求为空，表明等待写入的消息已经没有了，这时候文件缓存中的消息需要提交到磁盘，防止消息丢失；或者如果已经写入文件的大小大于maxTransferSize，则提交到磁盘  
81.                           //这里需要注意的是，会出现这样一种情况，刚好最后一个segment的文件快满了，这时候是不会roll出一个新的segment写入消息的，而是直接追加到原来的segment尾部，可能导致segment对应的文件大小大于配置的单个segment大小  
82.                         if (req == null || last.fileMessageSet.getSizeInBytes() > lastFlushPos + this.maxTransferSize) {  
83.                             // 强制force，确保内容保存到磁盘  
84.                             last.fileMessageSet.flush();  
85.                             lastFlushPos = last.fileMessageSet.highWaterMark();  
86.                             // 通知回调  
87.                             //异步写入比组写入可靠，因为异步写入一定是提交到磁盘的时候才进行回调的，而组写入如果依赖组提交的方式，则可能会丢失数据，因为组写入在消息写入到文件缓存的时候就进行回调了(除非设置unflushThreshold=1)  
88.                             for (final WriteRequest request : toFlush) {  
89.                                 request.cb.appendComplete(request.result);  
90.                             }  
91.                             toFlush.clear();  
92.                             // 是否需要roll  
93.                             this.mayBeRoll();  
94.                             // 如果切换文件，重新获取last  
95.                             if (this.segments.last() != last) {  
96.                                 last = null;  
97.                             }  
98.                             continue;  
99.                         }  
100.                     }  
101.                 }  
102.   
103.                 if (req == null) {  
104.                     continue;  
105.                 }  
106.                 //写入文件，并计算写入位置  
107.                 final int remainning = req.buf.remaining();  
108.                 //写入位置为：当前segment给定的值 + 加上文件已有的长度   
109. final long offset = last.start + last.fileMessageSet.append(req.buf);  
110.                 req.result = Location.create(offset, remainning);  
111.                 if (req.cb != null) {  
112.                     toFlush.add(req);  
113.                 }  
114.                 req = null;  
115.             } catch (final IOException e) {  
116.                 log.error("Append message failed,*critical error*,the group commit thread would be terminated.", e);  
117.                 // TODO io异常没办法处理了，简单跳出?  
118.                 break;  
119.             } catch (final InterruptedException e) {  
120.                 // ignore  
121.             }  
122.         }  
123.         // terminated  
124.         //关闭store 前，将等待写入队列中的剩余消息写入最后一个文件，这时候如果最后一个Segment满了也不会roll出新的Segment，持续的将消息写入到最后一个Segment，所以这时候也会发生Segment的size大于配置的size的情况  
125.         try {  
126.             for (WriteRequest request : this.bufferQueue) {  
127.                 final int remainning = request.buf.remaining();  
128.                 final long offset = last.start + last.fileMessageSet.append(request.buf);  
129.                 if (request.cb != null) {  
130.                     request.cb.appendComplete(Location.create(offset, remainning));  
131.                 }  
132.             }  
133.             this.bufferQueue.clear();  
134.         } catch (IOException e) {  
135.             log.error("Append message failed", e);  
136.         }  
137.     }  
138. ……  
139.   
140.      //Append多个消息，返回写入的位置  
141.     public void append(final List<Long> msgIds, final List<PutCommand> putCmds, final AppendCallback cb) {  
142.         this.appendBuffer(MessageUtils.makeMessageBuffer(msgIds, putCmds), cb);  
143.     }  
144.   
145.     /** 
146.      * 重放事务操作，如果消息没有存储成功，则重新存储，并返回新的位置 
147.      */  
148.     public void replayAppend(final long offset, final int length, final int checksum, final List<Long> msgIds,  
149.             final List<PutCommand> reqs, final AppendCallback cb) throws IOException {  
150.        //如果消息没有存储，则重新存储，写到最后一个Segment尾部  
151.         final Segment segment = this.findSegment(this.segments.view(), offset);  
152.         if (segment == null) {  
153.             this.append(msgIds, reqs, cb);  
154.         } else {  
155.             final MessageSet messageSet = segment.fileMessageSet.slice(offset - segment.start, offset - segment.start + length);  
156.             final ByteBuffer buf = ByteBuffer.allocate(length);  
157.             messageSet.read(buf, offset - segment.start);  
158.             buf.flip();  
159.             final byte[] bytes = new byte[buf.remaining()];  
160.             buf.get(bytes);  
161.             // 这个校验和是整个消息的校验和，这跟message的校验和不一样，注意区分  
162.             final int checkSumInDisk = CheckSum.crc32(bytes);  
163.             // 没有存入，则重新存储  
164.             if (checksum != checkSumInDisk) {  
165.                 this.append(msgIds, reqs, cb);  
166.             } else {  
167.                 // 正常存储了消息，无需处理  
168.                 if (cb != null) {   
169.                     this.notifyCallback(cb, null);  
170.                 }  
171.             }  
172.         }  
173.     }  
174.   
175. //判断是否需要roll，如果当前 messagestore最后一个segment的size>=配置的segment size，则产生新的segment，并将新的segment作为最后一个segment，原来最后的segment提交一次，并将mutable设置为false  
176. private void mayBeRoll() throws IOException {  
177.         if (this.segments.last().fileMessageSet.getSizeInBytes() >= this.metaConfig.getMaxSegmentSize()) {  
178.             this.roll();  
179.         }  
180.     }  
181.   
182.     String nameFromOffset(final long offset) {  
183.         final NumberFormat nf = NumberFormat.getInstance();  
184.         nf.setMinimumIntegerDigits(20);  
185.         nf.setMaximumFractionDigits(0);  
186.         nf.setGroupingUsed(false);  
187.         return nf.format(offset) + FILE_SUFFIX;  
188.     }  
189.   
190. private long nextAppendOffset() throws IOException {  
191.         final Segment last = this.segments.last();  
192.         last.fileMessageSet.flush();  
193.         return last.start + last.size();  
194.     }  
195.   
196.     private void roll() throws IOException {  
197.         final long newOffset = this.nextAppendOffset();  
198.         //新的segment对应的存储文件的命名为原来最后一个segment的起始位置 + segment的size  
199.         final File newFile = new File(this.partitionDir, this.nameFromOffset(newOffset));  
200.         this.segments.last().fileMessageSet.flush();  
201.         this.segments.last().fileMessageSet.setMutable(false);  
202.         this.segments.append(new Segment(newOffset, newFile));  
203.     }  
204.   
205. //判断是否需要消息提交到磁盘，判断的条件有两个，如果达到组提交的条件或者达到间隔的提交时间  
206. private void mayBeFlush(final int numOfMessages) throws IOException {  
207.         if (this.unflushed.addAndGet(numOfMessages) > this.metaConfig.getTopicConfig(this.topic).getUnflushThreshold()  
208.                 || SystemTimer.currentTimeMillis() - this.lastFlushTime.get() > this.metaConfig.getTopicConfig(this.topic).getUnflushInterval()) {  
209.             this.flush0();  
210.         }  
211.     }  
212.   
213. //提交到磁盘  
214. public void flush() throws IOException {  
215.         this.writeLock.lock();  
216.         try {  
217.             this.flush0();  
218.         } finally {  
219.             this.writeLock.unlock();  
220.         }  
221.     }  
222.   
223.       
224. private void flush0() throws IOException {  
225.         if (this.useGroupCommit()) {  
226.             return;  
227.         }  
228.         //由于只有最后一个segment是可变，即可写入消息的，所以只需要提交最后一个segment的消息  
229.         this.segments.last().fileMessageSet.flush();  
230.         this.unflushed.set(0);  
231.         this.lastFlushTime.set(SystemTimer.currentTimeMillis());  
232.     }  
233.   
234.   
235. @Override  
236.     public void close() throws IOException {  
237.         this.closed = true;  
238.         this.interrupt();  
239.      //等待子线程完成写完异步队列中剩余未写的消息  
240.         try {  
241.             this.join(500);  
242.         } catch (InterruptedException e) {  
243.             Thread.currentThread().interrupt();  
244.         }  
245.       //关闭segment，保证内容都已经提交到磁盘  
246.         for (final Segment segment : this.segments.view()) {  
247.             segment.fileMessageSet.close();  
248.         }  
249.     }  
250.   
251. //返回segment的信息，主要包括segment的开始位置以及 segment 的size  
252. public List<SegmentInfo> getSegmentInfos() {  
253.         final List<SegmentInfo> rt = new ArrayList<SegmentInfo>();  
254.         for (final Segment seg : this.segments.view()) {  
255.             rt.add(new SegmentInfo(seg.start, seg.size()));  
256.         }  
257.         return rt;  
258.     }  
259.   
260. /** 
261.      * 返回当前最大可读的offset 
262.      */  
263.   //需要注意的是，在文件缓存中的消息是不可读的，可以通过getSizeInBytes（）方法来判断还有多少内容还在文件缓存中，getSizeInBytes()方法返回的值是包括所有在磁盘和缓存中的size  
264.     public long getMaxOffset() {  
265.         final Segment last = this.segments.last();  
266.         if (last != null) {  
267.             return last.start + last.size();  
268.         } else {  
269.             return 0;  
270.         }  
271.     }  
272.   
273.     /** 
274.      * 返回当前最小可读的offset 
275.      */  
276.     public long getMinOffset() {  
277.         Segment first = this.segments.first();  
278.         if (first != null) {  
279.             return first.start;  
280.         } else {  
281.             return 0;  
282.         }  
283.     }  
284.   
285. /** 
286.      * 根据offset和maxSize返回所在MessageSet, 当offset超过最大offset的时候返回null， 
287.      * 当offset小于最小offset的时候抛出ArrayIndexOutOfBounds异常 
288.      */  
289.   //代码的注释以及清楚的解析了作用  
290.     public MessageSet slice(final long offset, final int maxSize) throws IOException {  
291.         final Segment segment = this.findSegment(this.segments.view(), offset);  
292.         if (segment == null) {  
293.             return null;  
294.         } else {  
295.             return segment.fileMessageSet.slice(offset - segment.start, offset - segment.start + maxSize);  
296.         }  
297.     }  
298.   
299. /** 
300.      * 根据offset查找文件,如果超过尾部，则返回null，如果在头部之前，则抛出ArrayIndexOutOfBoundsException 
301.      */  
302.   //指定位置找到对应的segment，由于前面的文件组织方式，所以这里可以采用2分查找的方式，  
303.   //效率很高  
304.     Segment findSegment(final Segment[] segments, final long offset) {  
305.         if (segments == null || segments.length < 1) {  
306.             return null;  
307.         }  
308.         // 老的数据不存在，返回最近最老的数据  
309.         final Segment last = segments[segments.length - 1];  
310.         // 在头部以前，抛出异常  
311.         if (offset < segments[0].start) {  
312.             throw new ArrayIndexOutOfBoundsException();  
313.         }  
314.         // 刚好在尾部或者超出范围，返回null  
315.         if (offset >= last.start + last.size()) {  
316.             return null;  
317.         }  
318.         // 根据offset二分查找  
319.         int low = 0;  
320.         int high = segments.length - 1;  
321.         while (low <= high) {  
322.             final int mid = high + low >>> 1;  
323.             final Segment found = segments[mid];  
324.             if (found.contains(offset)) {  
325.                 return found;  
326.             } else if (offset < found.start) {  
327.                 high = mid - 1;  
328.             } else {  
329.                 low = mid + 1;  
330.             }  
331.         }  
332.         return null;  
333.     } 

前面，我们已经把Broker存储最重要的一个类具体分析了一遍，接下来，我们分析一下其删除的策略。前面介绍过Messagestore采用的多文件存储的组织方式，而存储空间不可能无限大，得有一定的删除策略对其进行删除以腾出空间给新的消息。

 

MetaQ允许自定义删除策略，需要实现接口DeletePolicy，默认提供了两种删除策略：过期删除（DiscardDeletePolicy）和过期打包删除（ArchiveDeletePolicy）。DiscardDeletePolicy和ArchiveDeletePolicy都比较简单，DiscardDeletePolicy主要是对于超过一定时期的文件进行删除，ArchiveDeletePolicy则是先打包备份再删除。

 

自定义策略是如何被识别和使用的呢，MetaQ定义了DeletePolicyFactory，所有删除策略的实例都由DeletePolicyFactory提供，DeletePolicyFactory对外提供了注册机制，利用反射机制生成实例，每个自定义的删除策略都必须有一个无参构造，DeletePolicyFactory生成实例代码如下:

 

Java代码  

1. public static DeletePolicy getDeletePolicy(String values) {  
2.         String[] tmps = values.split(",");  
3.         String name = tmps[0];  
4.         Class<? extends DeletePolicy> clazz = policyMap.get(name);  
5.         if (clazz == null) {  
6.             throw new UnknownDeletePolicyException(name);  
7.         }  
8.         try {  
9.            //直接调用class的newInstance()方法，该方法必须要求有一个无参构造  
10.             DeletePolicy deletePolicy = clazz.newInstance();  
11.             String[] initValues = null;  
12.             if (tmps.length >= 2) {  
13.                 initValues = new String[tmps.length - 1];  
14.                 System.arraycopy(tmps, 1, initValues, 0, tmps.length - 1);  
15.             }  
16.             deletePolicy.init(initValues);  
17.             return deletePolicy;  
18.         }  
19.         catch (Exception e) {  
20.             throw new MetamorphosisServerStartupException("New delete policy `" + name + "` failed", e);  
21.         }  
22.     }  

DeletePolicy和MessageStore如何结合在一起的呢？则是粘合剂MessageStoreManager，MessageStoreManager是存储模块的管家，负责与其他模块联系，也是MessageStore管理器，管理所有的MessageStore以及其删除策略，MessageStoreManager也是要好好分析的一个类。

 

 

Java代码  

1. private final ConcurrentHashMap<String/* topic */, ConcurrentHashMap<Integer/* partition */, MessageStore>> stores = new ConcurrentHashMap<String, ConcurrentHashMap<Integer, MessageStore>>();  
2. //前面的存储组织方式介绍过一个主题对应多一个分区，每个分区对应一个MessageStore实例，分区号使用数值来表示，stores就是按照该方式组织管理的  
3.     private final MetaConfig metaConfig;  
4. //参数配置  
5.     private ScheduledThreadPoolExecutor scheduledExecutorService;// =  
6.     // Executors.newScheduledThreadPool(2);  
7.   //调度服务，对不同的MessageStore实例flush，将数据提到到硬盘  
8.     private final DeletePolicy deletePolicy;  
9.   //删除策略选择器，这里采用的一个topic对应一种策略，而不是一个MessageStore对应一个策略实例，一个策略实例在同一个topic的不同MessageStore实例间是重用的  
10.     private DeletePolicySelector deletePolicySelector;  
11.     
12.     public static final int HALF_DAY = 1000 * 60 * 60 * 12;  
13.   //topic 集合  
14.     private final Set<Pattern> topicsPatSet = new HashSet<Pattern>();  
15.   
16.     private final ConcurrentHashMap<Integer, ScheduledFuture<?>> unflushIntervalMap = new ConcurrentHashMap<Integer, ScheduledFuture<?>>();  
17. //前面曾介绍过MessageStore的提交方式有两种：组提交和定时提交，unflushIntervalMap是存放  
18. //定时提交的任务  
19.     private Scheduler scheduler;  
20. //定时调度器，用于定时调度删除任务  
21.     public MessageStoreManager(final MetaConfig metaConfig, final DeletePolicy deletePolicy) {  
22.         this.metaConfig   = metaConfig;  
23.         this.deletePolicy = deletePolicy;  
24. //生成策略选择器  
25.         this.newDeletePolicySelector();  
26. //添加匿名监听器，监听topic列表变化，如果列表发生变化，则新增列表并重新生成选择器  
27.         this.metaConfig.addPropertyChangeListener("topics", new PropertyChangeListener() {  
28.             public void propertyChange(final PropertyChangeEvent evt) {  
29.                 MessageStoreManager.this.makeTopicsPatSet();  
30.                 MessageStoreManager.this.newDeletePolicySelector();  
31.             }  
32.         });  
33.  //添加匿名监听，监听unflushInternal变化，如果发生变化  
34.         this.metaConfig.addPropertyChangeListener("unflushInterval", new PropertyChangeListener() {  
35.             public void propertyChange(final PropertyChangeEvent evt) {  
36.                 MessageStoreManager.this.scheduleFlushTask();  
37.             }  
38.         });  
39.         this.makeTopicsPatSet();  
40.       //初始化调度  
41.         this.initScheduler();  
42.         // 定时flush，该方法作者有详细注释就不在解释了  
43.         this.scheduleFlushTask();  
44.     }  

 

MessageStoreManager实现接口Service，在启动是会调用init方法，关闭时调用dispose方法

Java代码  

1. public void init() {  
2.         // 加载已有数据并校验  
3.         try {  
4.             this.loadMessageStores(this.metaConfig);  
5.         } catch (final IOException e) {  
6.             log.error("load message stores failed", e);  
7.             throw new MetamorphosisServerStartupException("Initilize message store manager failed", e);  
8.         } catch (InterruptedException e) {  
9.             Thread.currentThread().interrupt();  
10.         }  
11.         this.startScheduleDeleteJobs();  
12.     }  
13.   
14. //  
15. private Set<File> getDataDirSet(final MetaConfig metaConfig) throws IOException {  
16.         final Set<String> paths = new HashSet<String>();  
17.         // public data path  
18.       //公共数据目录  
19.         paths.add(metaConfig.getDataPath());  
20.         // topic data path  
21.       //私有数据目录  
22.         for (final String topic : metaConfig.getTopics()) {  
23.             final TopicConfig topicConfig = metaConfig.getTopicConfig(topic);  
24.             if (topicConfig != null) {  
25.                 paths.add(topicConfig.getDataPath());  
26.             }  
27.         }  
28.         final Set<File> fileSet = new HashSet<File>();  
29.         for (final String path : paths) {  
30.             //验证数据目录是否存在  
31.             fileSet.add(this.getDataDir(path));  
32.         }  
33.         return fileSet;  
34.     }  
35.   
36. private void loadMessageStores(final MetaConfig metaConfig) throws IOException, InterruptedException {  
37. //加载数据目录列表，再加载每个目录下的数据  
38.         for (final File dir : this.getDataDirSet(metaConfig)) {  
39.             this.loadDataDir(metaConfig, dir);  
40.         }  
41.     }  
42.   
43.     private void loadDataDir(final MetaConfig metaConfig, final File dir) throws IOException, InterruptedException {  
44.         log.warn("Begin to scan data path:" + dir.getAbsolutePath());  
45.         final long start = System.currentTimeMillis();  
46.         final File[] ls = dir.listFiles();  
47.         int nThreads = Runtime.getRuntime().availableProcessors() + 1;  
48.         ExecutorService executor = Executors.newFixedThreadPool(nThreads);  
49.         int count = 0;  
50.       //将加载验证每个分区的数据包装成一个个任务  
51.         List<Callable<MessageStore>> tasks = new ArrayList<Callable<MessageStore>>();  
52.         for (final File subDir : ls) {  
53.             if (!subDir.isDirectory()) {  
54.                 log.warn("Ignore not directory path:" + subDir.getAbsolutePath());  
55.             } else {  
56.                 final String name = subDir.getName();  
57.                 final int index = name.lastIndexOf('-');  
58.                 if (index < 0) {  
59.                     log.warn("Ignore invlaid directory:" + subDir.getAbsolutePath());  
60.                     continue;  
61.                 }  
62.                   //包装任务  
63.                 tasks.add(new Callable<MessageStore>() {  
64.                     //回调方法，方法将具体的加载验证分区数据  
65. @Override  
66.                     public MessageStore call() throws Exception {  
67.                         log.warn("Loading data directory:" + subDir.getAbsolutePath() + "...");  
68.                         final String topic = name.substring(0, index);  
69.                         final int partition = Integer.parseInt(name.substring(index + 1));                   //构造MessageStore实例的时候会自动加载验证数据，在初始化MessageStore实例的时候会给该MessageStore实例选择该topic的删除策略  
70.                         final MessageStore messageStore = new MessageStore(topic, partition, metaConfig,  
71.                                 MessageStoreManager.this.deletePolicySelector.select(topic, MessageStoreManager.this.deletePolicy));  
72.                         return messageStore;  
73.                     }  
74.                 });  
75.                 count++;  
76.                 if (count % nThreads == 0 || count == ls.length) {  
77. //如果配置了并行加载，则使用并行加载  
78.                     if (metaConfig.isLoadMessageStoresInParallel()) {  
79.                         this.loadStoresInParallel(executor, tasks);  
80.                     } else {  
81. //串行加载验证数据  
82.                         this.loadStores(tasks);  
83.                     }  
84.                 }  
85.             }  
86.         }  
87.         executor.shutdownNow();  
88.         log.warn("End to scan data path in " + (System.currentTimeMillis() - start) / 1000 + " secs");  
89.     }  

 

在init方法中做的一件事情就是加载校验已有的数据，加载校验的方式有两种个，串行和并行。

Java代码  

1. //串行加载验证数据，则在主线程上完成验证加载工作，其缺点是较慢，好处是不会打乱日志顺序  
2. private void loadStores(List<Callable<MessageStore>> tasks) throws IOException, InterruptedException {  
3.         for (Callable<MessageStore> task : tasks) {  
4.             MessageStore messageStore;  
5.             try {  
6.                 messageStore = task.call();  
7.                 ConcurrentHashMap<Integer/* partition */, MessageStore> map = this.stores.get(messageStore.getTopic());  
8.                 if (map == null) {  
9.                     map = new ConcurrentHashMap<Integer, MessageStore>();  
10.                     this.stores.put(messageStore.getTopic(), map);  
11.                 }  
12.                 map.put(messageStore.getPartition(), messageStore);  
13.             } catch (IOException e) {  
14.                 throw e;  
15.             } catch (InterruptedException e) {  
16.                 throw e;  
17.             } catch (Exception e) {  
18.                 throw new IllegalStateException(e);  
19.             }  
20.         }  
21.         tasks.clear();  
22.     }  
23.   
24. //并行加载数据，当数据过多的时候，启动并行加载数据可以加快启动速度；但是会打乱启动的日志顺序，默认不启用。  
25. private void loadStoresInParallel(ExecutorService executor, List<Callable<MessageStore>> tasks) throws InterruptedException {  
26.         CompletionService<MessageStore> completionService = new ExecutorCompletionService<MessageStore>(executor);  
27.         for (Callable<MessageStore> task : tasks) {  
28.             completionService.submit(task);  
29.         }  
30.         for (int i = 0; i < tasks.size(); i++) {  
31.             try {  
32.                 //确保任务都已经运行完毕  
33.                 MessageStore messageStore = completionService.take().get();  
34.   
35.                 ConcurrentHashMap<Integer/* partition */, MessageStore> map = this.stores.get(messageStore.getTopic());  
36.                 if (map == null) {  
37.                     map = new ConcurrentHashMap<Integer, MessageStore>();  
38.                     this.stores.put(messageStore.getTopic(), map);  
39.                 }  
40.                 map.put(messageStore.getPartition(), messageStore);  
41.             } catch (ExecutionException e) {  
42.                 throw ThreadUtils.launderThrowable(e);  
43.             }  
44.         }  
45.         tasks.clear();  
46.     }  

 

MessageStoreManager关闭时调用dispose方法，确保资源都正确释放。

Java代码  

1. public void dispose() {  
2.  //关闭调度器和调度池  
3.         this.scheduledExecutorService.shutdown();  
4.         if (this.scheduler != null) {  
5.             try {  
6.                 this.scheduler.shutdown(true);  
7.             } catch (final SchedulerException e) {  
8.                 log.error("Shutdown quartz scheduler failed", e);  
9.             }  
10.         }  
11. //确保每一个 MessageStore实例都正确关闭  
12.         for (final ConcurrentHashMap<Integer/* partition */, MessageStore> subMap : MessageStoreManager.this.stores  
13.                 .values()) {  
14.             if (subMap != null) {  
15.                 for (final MessageStore msgStore : subMap.values()) {  
16.                     if (msgStore != null) {  
17.                         try {  
18.                             msgStore.close();  
19.                         } catch (final Throwable e) {  
20.                             log.error("Try to run close  " + msgStore.getTopic() + "," + msgStore.getPartition() + " failed", e);  
21.                         }  
22.                     }  
23.                 }  
24.             }  
25.         }  
26. //清空stores列表  
27.         this.stores.clear();  
28.     }  

 

MessageStoreManager对外提供了获取的MessageStore的方法getMessageStore(final String topic, final int partition)和getOrCreateMessageStore(final String topic, final int partition) throws IOException。

getMessageStore()从stores列表查找对应的MessageStore，如果不存在则返回空；而getOrCreateMessage()则先检查对应的topic是否曾经配置，如果没有则抛出异常，如果有则判断stores是否已有MessageStore实例，如果没有，则生成MessageStore实例放入到stores列表并返回，如果有，则直接返回。

Java代码  

1. public MessageStore getMessageStore(final String topic, final int partition) {  
2.         final ConcurrentHashMap<Integer/* partition */, MessageStore> map = this.stores.get(topic);  
3.         if (map == null) {  
4. //如果topic对应的MessageStore实例列表不存在，则直接返回null  
5.             return null;  
6.         }  
7.         return map.get(partition);  
8.     }  
9.   
10.     Collection<MessageStore> getMessageStoresByTopic(final String topic) {  
11.         final ConcurrentHashMap<Integer/* partition */, MessageStore> map = this.stores.get(topic);  
12.         if (map == null) {  
13.             return Collections.emptyList();  
14.         }  
15.         return map.values();  
16.     }  
17.   
18.     public MessageStore getOrCreateMessageStore(final String topic, final int partition) throws IOException {  
19.         return this.getOrCreateMessageStoreInner(topic, partition, 0);  
20.     }  
21.   
22.     public MessageStore getOrCreateMessageStore(final String topic, final int partition, final long offsetIfCreate) throws IOException {  
23.         return this.getOrCreateMessageStoreInner(topic, partition, offsetIfCreate);  
24.     }  
25.   
26.     private MessageStore getOrCreateMessageStoreInner(final String topic, final int partition, final long offsetIfCreate) throws IOException {  
27.       //判断topic是否可用，即是否在topicsPatSet列表中  
28.         if (!this.isLegalTopic(topic)) {  
29.             throw new IllegalTopicException("The server do not accept topic " + topic);  
30.         }  
31. //判断分区号是否正确  
32.         if (partition < 0 || partition >= this.getNumPartitions(topic)) {  
33.             log.warn("Wrong partition " + partition + ",valid partitions (0," + (this.getNumPartitions(topic) - 1) + ")");  
34.             throw new WrongPartitionException("wrong partition " + partition);  
35.         }  
36.         ConcurrentHashMap<Integer/* partition */, MessageStore> map = this.stores.get(topic);  
37. //如果topic对应的列表不存在，则生成列表，放进stores中  
38.         if (map == null) {  
39.             map = new ConcurrentHashMap<Integer, MessageStore>();  
40.             final ConcurrentHashMap<Integer/* partition */, MessageStore> oldMap = this.stores.putIfAbsent(topic, map);  
41.             if (oldMap != null) {  
42.                 map = oldMap;  
43.             }  
44.         }  
45. //判断列表中是否有存在分区号位partition为的MessageStore实例，如果有，直接返回；如果没有，则生成实例并放进列表中  
46.         MessageStore messageStore = map.get(partition);  
47.         if (messageStore != null) {  
48.             return messageStore;  
49.         } else {  
50.             // 对string加锁，特例  
51.             synchronized (topic.intern()) {  
52.                 messageStore = map.get(partition);  
53.                 // double check  
54.                 if (messageStore != null) {  
55.                     return messageStore;  
56.                 }  
57.                 messageStore = new MessageStore(topic, partition, this.metaConfig, this.deletePolicySelector.select(topic, this.deletePolicy), offsetIfCreate);  
58.                 log.info("Created a new message storage for topic=" + topic + ",partition=" + partition);  
59.                 map.put(partition, messageStore);  
60.             }  
61.         }  
62.         return messageStore;  
63.     }  
64.   
65.     boolean isLegalTopic(final String topic) {  
66.         for (final Pattern pat : this.topicsPatSet) {  
67.             if (pat.matcher(topic).matches()) {  
68.                 return true;  
69.             }  
70.         }  
71.         return false;  
72.     }  

通过MessageStoreManager，我们把MessageStore和删除策略很好的组织在一起，并在MessageStoreManager提供定时提交的功能，提升了数据的可靠性；通过MessageStoreManager也为其他模块访问存储模块提供了接口。

 

我觉得MessageStoreManager设计不好的地方在于topicsPatSet，在topic列表发生变化的时候，没有先清空topicsPatSet，而是直接添加，而且没有对topic对应的MessageStore实例进行重新初始化，如果MessageStore实例已经存在，新删除策略配置不能生效。个人建议是一旦topic列表发生变化的时候，重新初始化整个存储模块，保证一致性。

 

 Broker接收从Producer(Client端)发送的消息，也能够返回消息到Consumer(Client)，对于Broker来说，就是网络输入输出流的处理。

 

Broker使用淘宝内部的gecko框架作为网络传输框架，gecko是一个NIO框架，能够支持一下特性：

1、 可自定义协议,协议可扩展、紧凑、高效

2、 可自动管理重连，重连由客户端发起

3、 需进行心跳检测，及时发现连接失效

4、 请求应答模型应当支持同步和异步 

5、 连接的分组管理，并且在重连情况下能正确处理连接的分组

6、 请求的发送应当支持四种模型：  (1) 向单个连接发起请求  (2) 向分组内的某个连接发起请求，这个选择策略可定义 (3) 向分组内的所有连接发起请求  (4) 向多个分组发起请求，每个分组的请求遵循（2） 

7、 编程模型上尽量做到简单、易用、透明，向上层代码屏蔽网络处理的复杂细节。

8、 高度可配置，包括网络参数、服务层参数等 

9、 高度可靠，正确处理网络异常，对内存泄露等隐患有预防措施

10、 可扩展

如果时间允许的话，笔者也可以做一下 gecko的源码分析

 

由于网络模块与其他模块关联性极强，不像存储模块可以独立分析，所以此篇文章开始将从全局开始分析Broker。

 

 

先看看Broker的启动类MetamorphosisStartup：

 

Java代码  

1. public static void main(final String[] args) {  
2.     final String configFilePath = getConfigFilePath(args);  
3.     final MetaConfig metaConfig = getMetaConfig(configFilePath);  
4.     final MetaMorphosisBroker server = new MetaMorphosisBroker(metaConfig);  
5.     server.start();  
6. }  

从MetamorphosisStartup可以看出其逻辑是先加载了配置文件，然后构造了MetaMorphosisBroker实例，并调用该实例的start方法，MetaMorphosisBroker才是Broker真正的启动类。

 

 

看看真正的启动类MetaMorphosisBroker, MetaMorphosisBroker实现接口MetaMorphosisBrokerMBean，可以通过 jmx 协议关闭MetaMorphosisBroker。看看在构造MetaMorphosisBroker实例的时候干了些什么事情。

 

Java代码  

1. public MetaMorphosisBroker(final MetaConfig metaConfig) {  
2.         //配置信息  
3.     this.metaConfig = metaConfig;  
4.         //Broker对外提供的nio Server  
5.     this.remotingServer = newRemotingServer(metaConfig);  
6.        //线程池管理器，主要是提供给nio Server在并发环境下可以使用多线程处理，提高性能  
7.         this.executorsManager = new ExecutorsManager(metaConfig);  
8.        //全局唯一的id生成器       
9.         this.idWorker = new IdWorker(metaConfig.getBrokerId());  
10.        //存储模块管理器  
11.         this.storeManager = new MessageStoreManager(metaConfig, this.newDeletePolicy(metaConfig));  
12.        //统计模块管理器  
13.     this.statsManager = new StatsManager(this.metaConfig, this.storeManager, this.remotingServer);  
14.       //zookeeper客户端，前面介绍过metaq使用zookeeper作为中间协调者，Broker会将自己注册到zookeeper上，也会从zookeeper查询相关数据  
15.     this.brokerZooKeeper = new BrokerZooKeeper(metaConfig);  
16.       //网络输入输出流处理器        
17.         final BrokerCommandProcessor next = new BrokerCommandProcessor(this.storeManager, this.executorsManager,          this.statsManager, this.remotingServer, metaConfig, this.idWorker, this.brokerZooKeeper);  
18.      //事务存储引擎  
19.     JournalTransactionStore transactionStore = null;  
20.     try {  
21.         transactionStore = new JournalTransactionStore(metaConfig.getDataLogPath(), this.storeManager, metaConfig);  
22.     } catch (final Exception e) {  
23.         throw new MetamorphosisServerStartupException("Initializing transaction store failed.", e);  
24.     }  
25.       //带事务处理的网络输入输出流处理器，设计采用了责任链的设计模式，使用事务存储引擎存储中间结果  
26.     this.brokerProcessor = new TransactionalCommandProcessor(metaConfig, this.storeManager, this.idWorker, next, transactionStore, this.statsManager);  
27.       //钩子，JVM退出钩子，钩子实现在JVM退出的时候尽力正确关闭  MetaMorphosisBroker      
28.         this.shutdownHook = new ShutdownHook();  
29.       //注册钩子  
30.     Runtime.getRuntime().addShutdownHook(this.shutdownHook);  
31.      //注册MBean，因为MetaMorphosisBroker实现MetaMorphosisBrokerMBean接口，可以将自己作为MBean注册到MBeanServer  
32.     MetaMBeanServer.registMBean(this, null);  
33. }  

 

前面我们知道在启动的时候会调用MetaMorphosisBroker的start() 方法，来看看start()方法里究竟做了些什么事情

 

Java代码  

1. public synchronized void start() {  
2. //判断是否已经启动，如果已经启动，则不在启动  
3.     if (!this.shutdown) {  
4.         return;  
5.     }  
6.     this.shutdown = false;  
7. //初始化存储模块，加载验证已有数据  
8.     this.storeManager.init();  
9. //初始化线程池  
10.     this.executorsManager.init();  
11. //初始化统计模块  
12.     this.statsManager.init();  
13. //向nio server注册处理器  
14.     this.registerProcessors();  
15.     try {  
16. //NIO server启动   
17.         this.remotingServer.start();  
18.     } catch (final NotifyRemotingException e) {  
19.         throw new MetamorphosisServerStartupException("start remoting server failed", e);  
20.     }  
21.     try {  
22. //在/brokers/ids下创建临时节点，名称为节点Id  
23.         this.brokerZooKeeper.registerBrokerInZk();  
24. //如果为master节点，则创建/brokers/ids/master_config_checksum节点        
25.                 this.brokerZooKeeper.registerMasterConfigFileChecksumInZk();  
26. //添加主题列表监听器，监听主题列表变化，如果主题列表发生变化，则向zookeeper重新注册主题和分区信息            
27.                this.addTopicsChangeListener();  
28. //注册主题和分区信息  
29.             this.registerTopicsInZk();  
30. //设置标志位主题和分区注册成功  
31.         this.registerZkSuccess = true;  
32.     } catch (final Exception e) {  
33.         this.registerZkSuccess = false;  
34.         throw new MetamorphosisServerStartupException("Register broker to zk failed", e);  
35.     }  
36.     log.info("Starting metamorphosis server...");  
37. //初始化输入输出流处理器  
38.     this.brokerProcessor.init();  
39.     log.info("Start metamorphosis server successfully");  
40. }  

 

 

下面，让我们具体来看看start()方法里调用的MetaMorphosisBroker每一个方法，首先是registerProcessors()方法：

 

Java代码  

1. private void registerProcessors() {  
2. //注册Get命令处理器  
3. this.remotingServer.registerProcessor(GetCommand.class, new GetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));  
4. //注册Put命令的处理器         
5. this.remotingServer.registerProcessor(PutCommand.class, new PutProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));  
6. //查询最近有效的offset处理器  
7. this.remotingServer.registerProcessor(OffsetCommand.class, new OffsetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));  
8. //心跳检测处理器  
9. this.remotingServer.registerProcessor(HeartBeatRequestCommand.class, new VersionProcessor(this.brokerProcessor));  
10. //注册退出命令处理器  
11. this.remotingServer.registerProcessor(QuitCommand.class, new QuitProcessor(this.brokerProcessor));  
12. //注册统计信息查询处理器  
13. this.remotingServer.registerProcessor(StatsCommand.class, new StatsProcessor(this.brokerProcessor));  
14. //注册事务命令处理器  
15. this.remotingServer.registerProcessor(TransactionCommand.class, new TransactionProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));  
16. }  

 

 

依赖于不同的处理器，可以将不同的请求进行处理并返回结果。接下来就是addTopicsChangeListener()方法。

 

Java代码  

1. //addTopicsChangeListener方法比较简单，主要简单配置的topic列表的变化，前面介绍过MetaConfig提供监听机制监听topic列表的变化，该方法向MetaConfig注册一个匿名监听器监听topic列表变化，一旦发生变化则向zookeeper进行注册  
2. private void addTopicsChangeListener() {  
3.     // 监听topics列表变化并注册到zk  
4.     this.metaConfig.addPropertyChangeListener("topics", new PropertyChangeListener() {  
5.         public void propertyChange(final PropertyChangeEvent evt) {  
6.             try {  
7.                 MetaMorphosisBroker.this.registerTopicsInZk();  
8.             } catch (final Exception e) {  
9.                 log.error("Register topic in zk failed", e);  
10.             }  
11.         }  
12.     });  
13. }  

 

 

MetaMorphosisBroker在启动过程中被调用的方法还有registerTopicsInZk()方法，registerTopicsInZk完成向zookeeper注册topic和分区信息功能。在分析方法之前，有必要插入分析一下Broker在zk上注册的结构，代码在common工程的类MetaZookeeper，该结构是Broker和Client共享的。

 

Zk中有4中类型的根目录，分别是:

1) /consumers:存放消费者列表以及消费记录，消费者列表主要是以组的方式存在，结构主要如下：

       /consumers/xxGroup/ids/xxConsumerId：DATA(“：”后的DATA表示节点xxConsumerId对应的数据) 组内消费者Id;DATA为订阅主题列表，以”,”分隔

       /consumers/xxGroup/offsets/xxTopic/分区N：DATA  组内主题分区N的消费进度；DATA为topic下分区N具体进度值

       /consumers/xxGroup/owners/xxTopic/分区N：DATA 组内主题分区N的的消费者；DATA为消费者ID，表示XXTopic下分区N的数据由指定的消费者进行消费

 

2) /brokers/ids:存放Broker列表，如果Broker与Zookeeper失去连接，则会自动注销在/brokers/ids下的broker记录，例子如下：

    /brokers/ids/xxBroker

 

3) /brokers/topics-pub：存放发布的主题列表以及对应的可发送消息的Broker列表，例子如下：

    /brokers/topics-pub/xxTopic/xxBroker

    /brokers/topics-pub下记录的是可发送消息到xxTopic的Broker列表，意味着有多少个Broker允许存储Client发送到Topic数据

 

4) /brokers/topics-sub：存放订阅的主题列表以及对应可订阅的Broker列表，例子如下：

    /brokers/topics-sub/xxTopic/xxBroker

    /brokers/topics-sub下记录的可订阅xxTopic的Broker列表，意味着有多少个Broker允许被Client订阅topic的数据

具体代码如下：

 

Java代码  

1. public MetaZookeeper(final ZkClient zkClient, final String root) {  
2. //zk客户端  
3. this.zkClient = zkClient;  
4. //根路径，默认为空  
5. this.metaRoot = this.normalize(root);  
6. //前面讲的消费者列表  
7. this.consumersPath = this.metaRoot + "/consumers";  
8. //前面讲的brokers列表  
9. this.brokerIdsPath = this.metaRoot + "/brokers/ids";  
10. //前面讲的/brokers/topics-pub  
11. this.brokerTopicsPubPath = this.metaRoot + "/brokers/topics-pub";  
12. //前面讲的/brokers/topics-sub  
13. this.brokerTopicsSubPath = this.metaRoot + "/brokers/topics-sub";  
14. }  

 至于更复杂的，我们将在后面具体再进行分析，主要先了解该存储结构即可。

 

回归正题, registerTopicsInZk方法完成向zookeeper注册topic和分区信息功能

 

Java代码  

1. private void registerTopicsInZk() throws Exception {  
2.     // 先注册配置的topic到zookeeper  
3.     for (final String topic : this.metaConfig.getTopics()) {  
4.         this.brokerZooKeeper.registerTopicInZk(topic, true);  
5.     }  
6.     // 注册加载的topic到zookeeper  
7.         // 从下面代码可以看出，如果当前没有配置的topic，但前面配置过的topic如果有消息存在，依然会向zk注册，在某种程度，我认为这个设计不好，为什么？  
8. 答：我们前面分析过MessageStoreManager类，里面有getMessageStore()方法和getOrCreateMessageStore()方法，在调用getMessageStore()方法时没有检查参数topic是否在topicsPatSet列表中(topicsPatSet只包含了配置的topic)，而getOrCreateMessageStore()方法却检查了，这就意味着使用getOrCreateMessageStore()方法时，如果要查询获取不在topicsPatSet列表中的MessageStore实例会抛出异常，而调用getMessageStore()不会，让人产生疑惑。<span>个人见解认为一旦配置发生更改，如果要做热加载的话则先卸载再重新加载会更合适，而且在getOrCreateMessageStore()和getMessageStore()方法都使用topicsPatSet进行判断，保持一致性</span>  
9.     for (final String topic : this.storeManager.getMessageStores().keySet()) {  
10.         this.brokerZooKeeper.registerTopicInZk(topic, true);  
11.     }  
12. }  

 

MetaMorphosisBroker还有两个方法，一个是newDeletePolicy()方法，另一个是stop()方法。newDeletePolicy()用于生产全局的存储模块的删除策略，如果没有配置删除策略，则使用该策略。

 

Java代码  

1. //全局删除策略  
2. private DeletePolicy newDeletePolicy(final MetaConfig metaConfig) {  
3.     final String deletePolicy = metaConfig.getDeletePolicy();  
4.     if (deletePolicy != null) {  
5.         return DeletePolicyFactory.getDeletePolicy(deletePolicy);  
6.     }  
7.     return null;  
8. }  

 

 

 

而stop()方法则主要在MetaMorphosisBroker关闭的时候销毁资源，尽力保证MetaQ的正确关闭。

 

Java代码  

1. public synchronized void stop() {  
2. //如果关闭了，则不再关闭  
3.     if (this.shutdown) {  
4.         return;  
5.     }  
6.     log.info("Stopping metamorphosis server...");  
7.     this.shutdown = true;  
8. //关闭与zk连接，注销与当前节点相关的配置  
9.     this.brokerZooKeeper.close(this.registerZkSuccess);  
10.     try {  
11.         // Waiting for zookeeper to notify clients.  
12.         Thread.sleep(this.brokerZooKeeper.getZkConfig().zkSyncTimeMs);  
13.     } catch (InterruptedException e) {  
14.         // ignore  
15.     }  
16. //释放线程池  
17.     this.executorsManager.dispose();  
18. //释放存储模块  
19.     this.storeManager.dispose();  
20. //释放统计模块  
21.     this.statsManager.dispose();  
22. //关闭NIO Server  
23.     try {  
24.         this.remotingServer.stop();  
25.     } catch (final NotifyRemotingException e) {  
26.         log.error("Shutdown remoting server failed", e);  
27.     }  
28. //释放输入输出流处理器  
29.     this.brokerProcessor.dispose();  
30. //如果是独立的zk，则关闭zk  
31.     EmbedZookeeperServer.getInstance().stop();  
32. //释放钩子    
33.     if (!this.runShutdownHook && this.shutdownHook != null) {  
34.         Runtime.getRuntime().removeShutdownHook(this.shutdownHook);  
35.     }  
36.     log.info("Stop metamorphosis server successfully");  
37. }  

 前面介绍过MetaQ使用gecko框架作为网络传输框架，Gecko采用请求/响应的方式组织传输。MetaQ依据定义了请求和响应的命令，由于命令Client和Broker均需要使用，所以放在了common工程的类MetaEncodeCommand中：

Java代码  

1. public String GET_CMD = "get";  //请求数据请求  
2. public String RESULT_CMD = "result"; //结果响应(不包括消息)  
3. public String OFFSET_CMD = "offset"; //查询最近有效的offset请求  
4. public String PUT_CMD  = "put";  //发送消息命令请求  
5. public String SYNC_CMD = "sync"; //同步数据请求  
6. public String QUIT_CMD = "quit";  //退出请求，客户端发送此命令后，服务器将主动关闭连接  
7. public String VERSION_CMD = "version"; //查询服务器版本请求，也用于心跳检测  
8. public String STATS_CMD = "stats"; //查询统计信息请求  
9. public String TRANS_CMD = "transaction"; //事务请求  
10. //还有一个响应，这里没有响应头的定义，就是返回的消息集  

在分析Broker启动类MetaMorphosisBroker时，分析过registerProcessors()方法，针对于不同的请求，Broker注册了不同的处理器，详情见registerProcessors()方法。MetaQ传输采用文本协议设计，非常透明，MetaEncodeCommand定义是请求类型。

 

Broker分为请求和响应的命令，先看看请求的类图：

 

 

注：由于类图工具出现问题，AbstractRequestCommand跟RequestCommand的实现关系并画成依赖关系，请大家理解成实现关系，以后类图同样这样理解(除非接口与接口或者类与类关系使用依赖箭头一定是描述成依赖的)。

 

所有的请求的命令均继承AbstractRequestCommand类并实现RequestCommand接口，RequestCommand是Gecko框架定义的接口，所有的命令均在编码后能被Gecko框架组织传输，传输协议是透明的文本协议。AbstractRequestCommand定义了基本属性topic和opaque。

Java代码  

1. public abstract class AbstractRequestCommand implements RequestCommand, MetaEncodeCommand {  
2.     private Integer opaque; //主要用于标识请求，响应的时候该标识被带回，用于客户端区分是哪个请求的响应  
3.     private String topic;   
4. }  

 

 

 

响应命令的类图如下：

 

请求命令只分为两类，带有消息集合的响应（DataCommand）；带有其他结果的响应(BooleanCommand)。DataCommand里携带的消息的格式与消息的存储结构一直，这样可以提高Broker的处理能力，将消息解析、正确性等验证放在Client，充分发挥Client的计算能力。这里比较麻烦的一点就是其他结果的响应均有BooleanCommand完成，BooleanCommand中只有code和message熟悉，code用来返回响应状态码，比如统计结果的信息的携带就必须由message熟悉来完成，所以结果的响应能力有限，而且必须先转换成字符串，具体如下：

 

Java代码  

1. /** 
2.  * 应答命令,协议格式如下：</br> result code length opaque\r\n message 
3.  */  
4. public class BooleanCommand extends AbstractResponseCommand implements BooleanAckCommand {  
5.     private String message;  
6.     /** 
7.      * status code in http protocol 
8.      */  
9.     private final int code;//响应的状态码  
10.     public BooleanCommand(final int code, final String message, final Integer opaque) {  
11.         super(opaque);  
12.         this.code = code;  
13.         switch (this.code) {  
14.             case HttpStatus.Success:  
15.                 this.setResponseStatus(ResponseStatus.NO_ERROR);  
16.                 break;  
17.             default:  
18.                 this.setResponseStatus(ResponseStatus.ERROR);  
19.                 break;  
20.         }  
21.         this.message = message;  
22.     }  
23.   
24.     public String getErrorMsg() {  
25.         return this.message;  
26.     }  
27.   
28.     public int getCode() {  
29.         return this.code;  
30.     }  
31.   
32.     public void setErrorMsg(final String errorMsg) {  
33.         this.message = errorMsg;  
34.     }  
35.   
36.     public IoBuffer encode() {  
37. //对结果进行编码，以便能在网络上传输  
38.         final byte[] bytes = ByteUtils.getBytes(this.message);  
39.         final int messageLen = bytes == null ? 0 : bytes.length;  
40.         final IoBuffer buffer = IoBuffer.allocate(11 + ByteUtils.stringSize(this.code)  
41.                 + ByteUtils.stringSize(this.getOpaque()) + ByteUtils.stringSize(messageLen) + messageLen);  
42.         ByteUtils.setArguments(buffer, MetaEncodeCommand.RESULT_CMD, this.code, messageLen, this.getOpaque());  
43.         if (bytes != null) {  
44.             buffer.put(bytes);  
45.         }  
46.         buffer.flip();  
47.         return buffer;  
48.     }  
49. }  

 

前面讲到的每个请求都会携带一个属性opaque并且该opaque将会在响应里被带回，AbstractResponseCommand里定义了该被带回的属性opaque。

 

Java代码  

1. /** 
2.  * 应答命令基类 
3.  */  
4. public abstract class AbstractResponseCommand implements ResponseCommand, MetaEncodeCommand {  
5.     private Integer opaque;   
6.     private InetSocketAddress responseHost; // responseHost和responseTime尚未发现在哪来调用，预计是作者预留的属性  
7.     private long responseTime;  
8.     private ResponseStatus responseStatus; //响应状态  
9. }  

 

 

不知道研究过MetaQ源码的朋友们发现DataCommand的encode()是一个空实现没，虽然作者有注释，运行Broker确实不存在问题，但就从设计者的角度来看，还是有些小问题的，代码如下：

 

Java代码  

1. public class DataCommand extends AbstractResponseCommand {  
2.     private final byte[] data;  
3. ……  
4.     @Override  
5.     public IoBuffer encode() {  
6.     //作者注释： 不做任何事情，发送data command由transferTo替代  
7.        //笔者注释：因为Borker采用了BrokerCommandProcessor 中zeroCopy的机制，所以不会该encode方法的调用，但如果以后改动后，允许zeroCopy变成可配置项，如果该方法不实现，就会出现解析问题，因为配置容易加上，但容易忘记该处的实现。  
8.         return null;  
9.     }  
10. }  

 前面介绍了Broker在网络传输过程中使用的数据结构，同时也介绍了MetaQ使用了Gecko框架作为网络传输框架。

 

有人会问，Gecko什么调用MetaEncodeCommand的encode()方法，让命令变成可见的明文在网络传输，Gecko又在什么时候将网络传输的数据包装成一个个Command对象？

 

或许有人已经注意到了笔者在介绍Broker启动类MetaMorphosisBroker的时候估计漏掉了一个方法newRemotingServer()方法，即创建Gecko Server。

 

Java代码  

1. private static RemotingServer newRemotingServer(final MetaConfig metaConfig) {  
2.         final ServerConfig serverConfig = new ServerConfig();  
3.         serverConfig.setWireFormatType(new MetamorphosisWireFormatType()); //注册了MetamorphosisWireFormatType实例，该实例负责编码和解码Command  
4.         serverConfig.setPort(metaConfig.getServerPort());  
5.         final RemotingServer server = RemotingFactory.newRemotingServer(serverConfig);  
6.         return server;  
7. }  

 在该方法内注册了一个MetamorphosisWireFormatType实例，该实例负责Command 的编码解码工作，MetamorphosisWireFormatType实现接口WireFormatType。

 

 

Java代码  

1. public class MetamorphosisWireFormatType extends WireFormatType {  
2.     public static final String SCHEME = "meta";  
3.   
4.     public String getScheme() {  
5.         return SCHEME;  
6.     }  
7.   
8.     public String name() {  
9.         return "metamorphosis";  
10.     }  
11.   
12.     public CodecFactory newCodecFactory() {  
13.         return new MetaCodecFactory();  
14.     }  
15.   
16.     public CommandFactory newCommandFactory() {  
17.         return new MetaCommandFactory();  
18.     }  

MetamorphosisWireFormatType本身并没有进行编码解码，而是交给了类MetaCodecFactory去实现，另外我们也看到newCommandFactory()方法，该方法主要是用于连接的心跳检测。下面让我们分别来看看这两个类: MetaCommandFactory和MetaCodecFactory，MetaCommandFactory和MetaCodecFactory均是MetamorphosisWireFormatType的内部类

 

 

 

 

用于心跳检测的类MetaCommandFactory，该类主要有两个方法，创建心跳请求的createHeartBeatCommand()方法和响应心跳请求的createBooleanAckCommand()方法：

 

Java代码  

1. static class MetaCommandFactory implements CommandFactory {  
2.   
3.         public BooleanAckCommand createBooleanAckCommand(final CommandHeader request, final ResponseStatus responseStatus, final String errorMsg) {  
4. //响应心跳请求  
5.             int httpCode = -1;  
6.             switch (responseStatus) {  
7.                 case NO_ERROR:  
8.                     httpCode = HttpStatus.Success;  
9.                     break;  
10.                 case THREADPOOL_BUSY:  
11.                 case NO_PROCESSOR:  
12.                     httpCode = HttpStatus.ServiceUnavilable;  
13.                     break;  
14.                 case TIMEOUT:  
15.                     httpCode = HttpStatus.GatewayTimeout;  
16.                     break;  
17.                 default:  
18.                     httpCode = HttpStatus.InternalServerError;  
19.                     break;  
20.             }  
21.             return new BooleanCommand(httpCode, errorMsg, request.getOpaque());  
22.         }  
23.   
24.         public HeartBeatRequestCommand createHeartBeatCommand() {  
25. //前面介绍过VersionCommand用于心跳检测，就是用于此处  
26.             return new VersionCommand(OpaqueGenerator.getNextOpaque());  
27.         }  
28.     }  

 

 

MetaCodecFactory是MetaQ（包括Broker和Client，因为编码解码Broker和Client都需要）网络传输最重要的一个类，负责命令的编码解码，MetaCodecFactory要实现Gecko框架定义的接口CodecFactory，MetaCodecFactory实例才能被Gecko框架使用，接口CodecFactory就定义了两个方法，返回编码器和解码器（由于Client和Broker均需要使用到MetamorphosisWireFormatType，所以MetamorphosisWireFormatType放在common工程中）：

 

Java代码  

1. static class MetaCodecFactory implements CodecFactory {  
2.     //返回解码器  
3.         @Override  
4.         public Decoder getDecoder() {  
5.   
6.             return new Decoder() {  
7.                 //Gecko框架会在适当的时候调用该方法，并将数据放到参数buff中，  
8.                 //用户可以根据buff的内容进行解析，包装成对应的Command类型  
9.                 public Object decode(final IoBuffer buff, final Session session) {  
10.                     if (buff == null || !buff.hasRemaining()) {  
11.                         return null;  
12.                     }  
13.                     buff.mark();  
14.                      //匹配第一个{‘\r’, ‘\n’}，也就是找到命令的内容(不包括数据)，目前只有PutCommand和SynCommand有数据部分，其他的命令都只有命令的内容  
15.                     final int index = LINE_MATCHER.matchFirst(buff);  
16.                     if (index >= 0) {  
17.                           //获取命令内容  
18.                         final byte[] bytes = new byte[index - buff.position()];  
19.                         buff.get(bytes);  
20.                         //跳过\r\n  
21.                         buff.position(buff.position() + 2);  
22.                           //将命令字节数组转换成字符串  
23.                         final String line = ByteUtils.getString(bytes);  
24.                         if (log.isDebugEnabled()) {  
25.                             log.debug("Receive command:" + line);  
26.                         }  
27.                           //以空格为单位分离内容  
28.                         final String[] sa = SPLITER.split(line);  
29.                         if (sa == null || sa.length == 0) {  
30.                             throw new MetaCodecException("Blank command line.");  
31.                         }  
32.                           //判断内容的第一个字母  
33.                         final byte op = (byte) sa[0].charAt(0);  
34.                         switch (op) {  
35.                             case 'p':  
36.                                     //如果是p的话，认为是put命令，具体见MetaEncodeCommand定义的命令的内容并解析put命令，具体格式在每个命令的实现类里的注释都有，下面的各个方法的注释也有部分  
37.                                 return this.decodePut(buff, sa);  
38.                             case 'g':  
39.                                    //如果是g的话，认为是get命令  
40.                                 return this.decodeGet(sa);  
41.                             case 't':  
42.                                    //如果是g的话，认为是事务命令  
43.                                 return this.decodeTransaction(sa);  
44.                             case 'r':  
45.                                    //如果是g的话，认为是结果响应  
46.                                 return this.decodeBoolean(buff, sa);  
47.                             case 'v':  
48.                                      //如果是v的话，则可能是心跳请求或者数据响应，所以得使用更详细的信息进行判断  
49.                                 if (sa[0].equals("value")) {  
50.                                     return this.decodeData(buff, sa);  
51.                                 } else {  
52.                                     return this.decodeVersion(sa);  
53.                                 }  
54.                             case 's':  
55.                                //如果是s的话，则可能是统计请求或者同步，所以得使用更详细的信息进行判断  
56. if (sa[0].equals("stats")) {  
57.                                     return this.decodeStats(sa);  
58.                                 } else {  
59.                                     return this.decodeSync(buff, sa);  
60.                                 }  
61.                             case 'o':  
62.                                   //如果是o的话，查询最近可用位置请求  
63.                                 return this.decodeOffset(sa);  
64.                             case 'q':  
65.                                    //如果是q的话，退出连接请求  
66.                                 return this.decodeQuit();  
67.                             default:  
68.                                 throw new MetaCodecException("Unknow command:" + line);  
69.                         }  
70.                     } else {  
71.                         return null;  
72.                     }  
73.                 }  
74.   
75.                 private Object decodeQuit() {  
76.                     return new QuitCommand();  
77.                 }  
78.   
79.                 private Object decodeVersion(final String[] sa) {  
80.                     if (sa.length >= 2) {  
81.                         return new VersionCommand(Integer.parseInt(sa[1]));  
82.                     } else {  
83.                         return new VersionCommand(Integer.MAX_VALUE);  
84.                     }  
85.                 }  
86.   
87.                 // offset topic group partition offset opaque\r\n  
88.                 private Object decodeOffset(final String[] sa) {  
89.                     this.assertCommand(sa[0], "offset");  
90.                     return new OffsetCommand(sa[1], sa[2], Integer.parseInt(sa[3]), Long.parseLong(sa[4]), Integer.parseInt(sa[5]));  
91.                 }  
92.   
93.                 // stats item opaque\r\n  
94.                 // opaque可以为空  
95.                 private Object decodeStats(final String[] sa) {  
96.                     this.assertCommand(sa[0], "stats");  
97.                     int opaque = Integer.MAX_VALUE;  
98.                     if (sa.length >= 3) {  
99.                         opaque = Integer.parseInt(sa[2]);  
100.                     }  
101.                     String item = null;  
102.                     if (sa.length >= 2) {  
103.                         item = sa[1];  
104.                     }  
105.                     return new StatsCommand(opaque, item);  
106.                 }  
107.   
108.                 // value totalLen opaque\r\n data  
109.                 private Object decodeData(final IoBuffer buff, final String[] sa) {  
110.                     this.assertCommand(sa[0], "value");  
111.                     final int valueLen = Integer.parseInt(sa[1]);  
112.                     if (buff.remaining() < valueLen) {  
113.                         buff.reset();  
114.                         return null;  
115.                     } else {  
116.                         final byte[] data = new byte[valueLen];  
117.                         buff.get(data);  
118.                         return new DataCommand(data, Integer.parseInt(sa[2]));  
119.                     }  
120.                 }  
121.   
122.                 /** 
123.                  * result code length opaque\r\n message 
124.                  *  
125.                  * @param buff 
126.                  * @param sa 
127.                  * @return 
128.                  */  
129.                 private Object decodeBoolean(final IoBuffer buff, final String[] sa) {  
130.                     this.assertCommand(sa[0], "result");  
131.                     final int valueLen = Integer.parseInt(sa[2]);  
132.                     if (valueLen == 0) {  
133.                         return new BooleanCommand(Integer.parseInt(sa[1]), null, Integer.parseInt(sa[3]));  
134.                     } else {  
135.                         if (buff.remaining() < valueLen) {  
136.                             buff.reset();  
137.                             return null;  
138.                         } else {  
139.                             final byte[] data = new byte[valueLen];  
140.                             buff.get(data);  
141.                             return new BooleanCommand(Integer.parseInt(sa[1]), ByteUtils.getString(data), Integer.parseInt(sa[3]));  
142.                         }  
143.                     }  
144.                 }  
145.   
146.                 // get topic group partition offset maxSize opaque\r\n  
147.                 private Object decodeGet(final String[] sa) {  
148.                     this.assertCommand(sa[0], "get");  
149.                     return new GetCommand(sa[1], sa[2], Integer.parseInt(sa[3]), Long.parseLong(sa[4]), Integer.parseInt(sa[5]), Integer.parseInt(sa[6]));  
150.                 }  
151.   
152.                 // transaction key sessionId type [timeout] [unique qualifier]  
153.                 // opaque\r\n  
154.                 private Object decodeTransaction(final String[] sa) {  
155.                     this.assertCommand(sa[0], "transaction");  
156.                     final TransactionId transactionId = this.getTransactionId(sa[1]);  
157.                     final TransactionType type = TransactionType.valueOf(sa[3]);  
158.                     switch (sa.length) {  
159.                         case 7:  
160.                             // Both include timeout and unique qualifier.  
161.                             int timeout = Integer.valueOf(sa[4]);  
162.                             String uniqueQualifier = sa[5];  
163.                             TransactionInfo info = new TransactionInfo(transactionId, sa[2], type, uniqueQualifier, timeout);  
164.                             return new TransactionCommand(info, Integer.parseInt(sa[6]));  
165.                         case 6:  
166.                             // Maybe timeout or unique qualifier  
167.                             if (StringUtils.isNumeric(sa[4])) {  
168.                                 timeout = Integer.valueOf(sa[4]);  
169.                                 info = new TransactionInfo(transactionId, sa[2], type, null, timeout);  
170.                                 return new TransactionCommand(info, Integer.parseInt(sa[5]));  
171.                             } else {  
172.                                 uniqueQualifier = sa[4];  
173.                                 info = new TransactionInfo(transactionId, sa[2], type, uniqueQualifier, 0);  
174.                                 return new TransactionCommand(info, Integer.parseInt(sa[5]));  
175.                             }  
176.                         case 5:  
177.                             // Without timeout and unique qualifier.  
178.                             info = new TransactionInfo(transactionId, sa[2], type, null);  
179.                             return new TransactionCommand(info, Integer.parseInt(sa[4]));  
180.                         default:  
181.                             throw new MetaCodecException("Invalid transaction command:" + StringUtils.join(sa));  
182.                     }  
183.                 }  
184.   
185.                 private TransactionId getTransactionId(final String s) {  
186.                     return TransactionId.valueOf(s);  
187.                 }  
188.   
189.                 // sync topic partition value-length flag msgId  
190.                 // opaque\r\n  
191.                 private Object decodeSync(final IoBuffer buff, final String[] sa) {  
192.                     this.assertCommand(sa[0], "sync");  
193.                     final int valueLen = Integer.parseInt(sa[3]);  
194.                     if (buff.remaining() < valueLen) {  
195.                         buff.reset();  
196.                         return null;  
197.                     } else {  
198.                         final byte[] data = new byte[valueLen];  
199.                         buff.get(data);  
200.                         switch (sa.length) {  
201.                             case 7:  
202.                                 // old master before 1.4.4  
203.                                 return new SyncCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Long.valueOf(sa[5]), -1, Integer.parseInt(sa[6]));  
204.                             case 8:  
205.                                 // new master since 1.4.4  
206.                                 return new SyncCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Long.valueOf(sa[5]), Integer.parseInt(sa[6]), Integer.parseInt(sa[7]));  
207.                             default:  
208.                                 throw new MetaCodecException("Invalid Sync command:" + StringUtils.join(sa));  
209.                         }  
210.                     }  
211.                 }  
212.   
213.                 // put topic partition value-length flag checksum  
214.                 // [transactionKey]  
215.                 // opaque\r\n  
216.                 private Object decodePut(final IoBuffer buff, final String[] sa) {  
217.                     this.assertCommand(sa[0], "put");  
218.                     final int valueLen = Integer.parseInt(sa[3]);  
219.                     if (buff.remaining() < valueLen) {  
220.                         buff.reset();  
221.                         return null;  
222.                     } else {  
223.                         final byte[] data = new byte[valueLen];  
224.                         buff.get(data);  
225.                         switch (sa.length) {  
226.                             case 6:  
227.                                 // old clients before 1.4.4  
228.                                 return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, null, Integer.parseInt(sa[4]), Integer.parseInt(sa[5]));  
229.                             case 7:  
230.                                 // either transaction command or new clients since  
231.                                 // 1.4.4  
232.                                 String slot = sa[5];  
233.                                 char firstChar = slot.charAt(0);  
234.                                 if (Character.isDigit(firstChar) || '-' == firstChar) {  
235.                                     // slot is checksum.  
236.                                     int checkSum = Integer.parseInt(slot);  
237.                                     return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), checkSum, null, Integer.parseInt(sa[6]));  
238.                                 } else {  
239.                                     // slot is transaction id.  
240.                                     return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, this.getTransactionId(slot), Integer.parseInt(sa[4]), Integer.parseInt(sa[6]));  
241.                                 }  
242.                             case 8:  
243.                                 // New clients since 1.4.4  
244.                                 // A transaction command  
245.                                 return new PutCommand(sa[1], Integer.parseInt(sa[2]), data, Integer.parseInt(sa[4]), Integer.parseInt(sa[5]), this.getTransactionId(sa[6]), Integer.parseInt(sa[7]));  
246.                             default:  
247.                                 throw new MetaCodecException("Invalid put command:" + StringUtils.join(sa));  
248.                         }  
249.                     }  
250.                 }  
251.   
252.                 private void assertCommand(final String cmd, final String expect) {  
253.                     if (!expect.equals(cmd)) {  
254.                         throw new MetaCodecException("Expect " + expect + ",but was " + cmd);  
255.                     }  
256.                 }  
257.             };  
258.         }  
259.   
260.         @Override  
261.         public Encoder getEncoder() {  
262.             //返回编码器  
263. return new Encoder() {  
264.                 @Override  
265.                 public IoBuffer encode(final Object message, final Session session) {  
266.                       //框架会在适当的时候调用编码器的encode()方法，前面说过如果响应的命令是DataCommand的时候假设不是zeroCopy的话，会出现问题。原因就在这里，因为如果不使用zeroCopy的话，返回给Gecko框架的是一个DataCommand的实例，这时候会调用到此方法，而此方法并没有按照DataCommand的格式进行编码，解码器会识别不了，所以容易出问题  
267.                     return ((MetaEncodeCommand) message).encode();  
268.                 }  
269.             };  
270.         }  

 

 

前面还介绍到过MetaMorphosisBroker在启动时会注册请求类型与Processor的映射，见代码：

 

Java代码  

1. private void registerProcessors() {  
2.         this.remotingServer.registerProcessor(GetCommand.class, new GetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));  
3.         this.remotingServer.registerProcessor(PutCommand.class, new PutProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));  
4.         this.remotingServer.registerProcessor(OffsetCommand.class, new OffsetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));  
5.         this.remotingServer.registerProcessor(HeartBeatRequestCommand.class, new VersionProcessor(this.brokerProcessor));  
6.         this.remotingServer.registerProcessor(QuitCommand.class, new QuitProcessor(this.brokerProcessor));  
7.         this.remotingServer.registerProcessor(StatsCommand.class, new StatsProcessor(this.brokerProcessor));  
8.         this.remotingServer.registerProcessor(TransactionCommand.class, new TransactionProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));  
9. }  

依据注册的类型，Gecko框架将会根据解析出来的命令实例调用处理器的不同方法，并返回不同请求的响应。下面让我们来看看不同的处理到底做了些什么事情？因为是Broker针对请求的处理，所以所有的Processor都在server工程中，先上类图：

 

 

所有的处理器均实现了RequestProcessor接口，该接口由Gecko框架定义，RequestProcessor类中只定义了两个方法：

Java代码  

1. public interface RequestProcessor<T extends RequestCommand> {  
2.     /** 
3.      * 处理请求 
4.      *  
5.      * @param request请求命令 
6.      * @param conn 请求来源的连接 
7.      */  
8.     public void handleRequest(T request, Connection conn);  
9.   
10.   
11.     /** 
12.      * 用户自定义的线程池，如果提供，那么请求的处理都将在该线程池内执行 
13.      *  
14.      * @return 
15.      */  
16.     public ThreadPoolExecutor getExecutor();  
17. }  

 

所以，加上上一篇文章，我们可以得出MetaQ的大致网络处理流程图解如下：

 

 

 上一篇以及上上篇基本介绍了MetaQ如何使用Gecko框架在网络上传输数据，今天将继续进一步介绍在Broker，各种命令的处理逻辑（暂时将不涉及到事务处理）。

 

依旧是在MetaMorphosisBroker的registerProcessors()方法中，我们可以注意到一点，每个Processor的实例在构造的时候都注入了一个brokerProcessor的变量，该变量的类型为CommandProcessor。其实，各个Processor的业务逻辑又委托给了CommandProcessor进行处理，比如我们看看其中的GetProcessor的源码：

Java代码  

1. public class GetProcessor implements RequestProcessor<GetCommand> {  
2.     public static final Logger log = LoggerFactory.getLogger(GetProcessor.class);  
3.   
4.     private final ThreadPoolExecutor executor;  
5.   
6.     private final CommandProcessor processor;  
7.   
8.     public GetProcessor(final CommandProcessor processor, final ThreadPoolExecutor executor) {  
9.         this.processor = processor;  
10.         this.executor = executor;  
11.     }  
12.   
13.     @Override  
14.     public ThreadPoolExecutor getExecutor() {  
15.         return this.executor;  
16.     }  
17.   
18.     @Override  
19.     public void handleRequest(final GetCommand request, final Connection conn) {  
20.         // Processor并没有处理具体的业务逻辑，而是将业务逻辑交给CommandProcessor的processGetCommand()进行处理，Processor只是将处理结果简单的返回给客户端  
21. final ResponseCommand response = this.processor.processGetCommand(request, SessionContextHolder.getOrCreateSessionContext(conn, null));  
22.         if (response != null) {  
23.             RemotingUtils.response(conn, response);  
24.         }  
25.     }  
26. }  

 

CommandProcessor业务逻辑的处理模块采用责任链的处理方式，目前来说只有两个类型的业务逻辑处理单元：带有事务处理(TransactionalCommandProcessor)的和不带有事务处理(BrokerCommandProcessor)的。*惯，先上类图：

 

 

CommandProcessor接口定义如下：

Java代码  

1. public interface CommandProcessor extends Service {  
2.     //处理Put命令，结果通过PutCallback的回调返回  
3.     public void processPutCommand(final PutCommand request, final SessionContext sessionContext, final PutCallback cb) throws Exception;  
4.      //处理Get命令  
5.     public ResponseCommand processGetCommand(GetCommand request, final SessionContext ctx);  
6.   
7.     /** 
8.      * Under conditions that cannot use notify-remoting directly. 
9.      */  
10.      //处理Get命令，并根据条件zeroCopy是否使用zeroCopy  
11.     public ResponseCommand processGetCommand(GetCommand request, final SessionContext ctx, final boolean zeroCopy);  
12.     //处理查询最近可用offset位置请求  
13.     public ResponseCommand processOffsetCommand(OffsetCommand request, final SessionContext ctx);  
14.     //处理退出请求  
15.     public void processQuitCommand(QuitCommand request, final SessionContext ctx);  
16.      
17.     public ResponseCommand processVesionCommand(VersionCommand request, final SessionContext ctx);  
18.     //处理统计请求  
19.     public ResponseCommand processStatCommand(StatsCommand request, final SessionContext ctx);  
20.     //下面主要定义与事务相关的方法，暂时先不介绍  
21.     public void removeTransaction(final XATransactionId xid);  
22.   
23.     public Transaction getTransaction(final SessionContext context, final TransactionId xid) throws MetamorphosisException, XAException;  
24.   
25.     public void forgetTransaction(final SessionContext context, final TransactionId xid) throws Exception;  
26.   
27.     public void rollbackTransaction(final SessionContext context, final TransactionId xid) throws Exception;  
28.   
29.     public void commitTransaction(final SessionContext context, final TransactionId xid, final boolean onePhase) throws Exception;  
30.   
31.     public int prepareTransaction(final SessionContext context, final TransactionId xid) throws Exception;  
32.   
33.     public void beginTransaction(final SessionContext context, final TransactionId xid, final int seconds) throws Exception;  
34.   
35.     public TransactionId[] getPreparedTransactions(final SessionContext context, String uniqueQualifier) throws Exception;  
36. }  

 

细心的读者会发现，每个定义的方法的参数都有一个参数SessionContext，SessionContext携带了连接的信息，由Broker创建，具体代码见SessionContextHolder的getOrCreateSessionContext()方法，getOrCreateSessionContext()方法在Processor委托给CommandProcessor处理业务逻辑时被调用。

 

BrokerCommandProcessor和TransactionalCommandProcessor其实就是各模块的粘合剂，将各模块的功能统一协调形成整体对外提供功能。BrokerCommandProcessor的实现并不难理解，下面让我们来具体分析一下BrokerCommandProcessor这个类：

Java代码  

1. //Put请求的业务逻辑处理  
2. @Override  
3. public void processPutCommand(final PutCommand request, final SessionContext sessionContext, final PutCallback cb) {  
4.         final String partitionString = this.metaConfig.getBrokerId() + "-" + request.getPartition();  
5. //统计计算  
6.         this.statsManager.statsPut(request.getTopic(), partitionString, 1);  
7.         this.statsManager.statsMessageSize(request.getTopic(), request.getData().length);  
8.         int partition = -1;  
9.         try {  
10. //如果对应存储的分区已经关闭，则拒绝该消息  
11.             if (this.metaConfig.isClosedPartition(request.getTopic(), request.getPartition())) {  
12.                 log.warn("Can not put message to partition " + request.getPartition() + " for topic=" + request.getTopic() + ",it was closed");  
13.                 if (cb != null) {  
14.                     cb.putComplete(new BooleanCommand(HttpStatus.Forbidden, this.genErrorMessage(request.getTopic(), request.getPartition()) + "Detail:partition[" + partitionString + "] has been closed", request.getOpaque()));  
15.                 }  
16.                 return;  
17.             }  
18.   
19.             partition = this.getPartition(request);  
20. //获取对应Topic分区的MessageStore实例  
21.             final MessageStore store = this.storeManager.getOrCreateMessageStore(request.getTopic(), partition);  
22.             // 如果是动态添加的topic，需要注册到zk  
23. //就到目前为止，我着实没想明白下面这句代码的用途是什么？   
24. //如果topic没有在该Broker的配置中配置，在MessageStoreManager中的isLegalTopic()方法中检查就通不过而抛出异常，那么下面这句代码怎么样都不会被执行，而Client要向Broker发送消息，一定要先发布topic，保证topic在zk发布；   
25.             this.brokerZooKeeper.registerTopicInZk(request.getTopic(), false);  
26.             // 设置唯一id  
27.             final long messageId = this.idWorker.nextId();  
28.             //存储消息，之前的文章介绍过Broker的存储使用回调的方式，易于异步的实现，代码简单不分析  
29. store.append(messageId, request, new StoreAppendCallback(partition, partitionString, request, messageId, cb));  
30.         } catch (final Exception e) {  
31. //发生异常，统计计算回滚  
32.             this.statsManager.statsPutFailed(request.getTopic(), partitionString, 1);  
33.             log.error("Put message failed", e);  
34.             if (cb != null) {  
35. //返回结果  
36.                 cb.putComplete(new BooleanCommand(HttpStatus.InternalServerError, this.genErrorMessage(request.getTopic(), partition) + "Detail:" + e.getMessage(), request.getOpaque()));  
37.             }  
38.         }  
39.     }  
40.   
41. @Override  
42. // GET请求的业务逻辑处理  
43. public ResponseCommand processGetCommand(final GetCommand request, final SessionContext ctx) {  
44. //默认为zeroCopy  
45.         return this.processGetCommand(request, ctx, true);  
46.     }  
47.   
48.     @Override  
49.     public ResponseCommand processGetCommand(final GetCommand request, final SessionContext ctx, final boolean zeroCopy) {  
50. //获取查询信息  
51.         final String group = request.getGroup();  
52.         final String topic = request.getTopic();  
53. //统计计数（请求数统计）  
54.         this.statsManager.statsGet(topic, group, 1);  
55.   
56.         // 如果分区被关闭,禁止读数据 --wuhua  
57.         if (this.metaConfig.isClosedPartition(topic, request.getPartition())) {  
58.             log.warn("can not get message for topic=" + topic + " from partition " + request.getPartition() + ",it closed,");  
59.             return new BooleanCommand(HttpStatus.Forbidden, "Partition[" + this.metaConfig.getBrokerId() + "-" + request.getPartition() + "] has been closed", request.getOpaque());  
60.         }  
61. //获取topic对应分区的MessageStore实例，如果实例不存在，则返回NotFound  
62.         final MessageStore store = this.storeManager.getMessageStore(topic, request.getPartition());  
63.         if (store == null) {  
64. //统计计数  
65.             this.statsManager.statsGetMiss(topic, group, 1);  
66.             return new BooleanCommand(HttpStatus.NotFound, "The topic `" + topic + "` in partition `" + request.getPartition() + "` is not exists", request.getOpaque());  
67.         }  
68. //如果请求的起始位置<0，判定该请求无效  
69.         if (request.getMaxSize() <= 0) {  
70.             return new BooleanCommand(HttpStatus.BadRequest, "Bad request,invalid max size:" + request.getMaxSize(), request.getOpaque());  
71.         }  
72.         try {  
73. //读取由request.getOffset()开始的消息集合  
74.             final MessageSet set = store.slice(request.getOffset(), Math.min(this.metaConfig.getMaxTransferSize(), request.getMaxSize()));  
75. //如果当前消息集不为空  
76.             if (set != null) {  
77. //判断是否zeroCopy，如果是zeroCopy，则直接写；如果不是，则将消息集包装成DataCommand，这也就是前面为什么说DataCommand要实现encode()方法的缘故  
78.                 if (zeroCopy) {  
79.                     set.write(request, ctx);  
80.                     return null;  
81.                 } else {  
82.                     // refer to the code of line 440 in MessageStore  
83.                     // create two copies of byte array including the byteBuffer  
84.                     // and new bytes  
85.                     // this may not a good use case of Buffer  
86.                     final ByteBuffer byteBuffer = ByteBuffer.allocate(Math.min(this.metaConfig.getMaxTransferSize(), request.getMaxSize()));  
87.                     set.read(byteBuffer);  
88.                     byteBuffer.flip();  
89.                     final byte[] bytes = new byte[byteBuffer.remaining()];  
90.                     byteBuffer.get(bytes);  
91.                     return new DataCommand(bytes, request.getOpaque());  
92.                 }  
93.             } else {  
94. //如果为空消息集，则认为请求无效  
95. //统计计数  
96.                 this.statsManager.statsGetMiss(topic, group, 1);  
97.                 this.statsManager.statsGetFailed(topic, group, 1);  
98.   
99.                 // 当请求的偏移量大于实际最大值时,返回给客户端实际最大的偏移量.  
100.                 final long maxOffset = store.getMaxOffset();  
101.                 final long requestOffset = request.getOffset();  
102.                 if (requestOffset > maxOffset && (this.metaConfig.isUpdateConsumerOffsets() || requestOffset == Long.MAX_VALUE)) {  
103.                     log.info("offset[" + requestOffset + "] is exceeded,tell the client real max offset: " + maxOffset + ",topic=" + topic + ",group=" + group);  
104.                     this.statsManager.statsOffset(topic, group, 1);  
105.                     return new BooleanCommand(HttpStatus.Moved, String.valueOf(maxOffset), request.getOpaque());  
106.                 } else {  
107.                     return new BooleanCommand(HttpStatus.NotFound, "Could not find message at position " + requestOffset, request.getOpaque());  
108.                 }  
109.             }  
110.         } catch (final ArrayIndexOutOfBoundsException e) {  
111.             log.error("Could not get message from position " + request.getOffset() + ",it is out of bounds,topic=" + topic);  
112.             // 告知最近可用的offset  
113.             this.statsManager.statsGetMiss(topic, group, 1);  
114.             this.statsManager.statsGetFailed(topic, group, 1);  
115.             final long validOffset = store.getNearestOffset(request.getOffset());  
116.             this.statsManager.statsOffset(topic, group, 1);  
117.             return new BooleanCommand(HttpStatus.Moved, String.valueOf(validOffset), request.getOpaque());  
118.         } catch (final Throwable e) {  
119.             log.error("Could not get message from position " + request.getOffset(), e);  
120.             this.statsManager.statsGetFailed(topic, group, 1);  
121.             return new BooleanCommand(HttpStatus.InternalServerError, this.genErrorMessage(request.getTopic(), request.getPartition()) + "Detail:" + e.getMessage(), request.getOpaque());  
122.         }  
123.     }  
124.   
125. //查询最近可用offset请求的业务逻辑处理  
126. @Override  
127.     public ResponseCommand processOffsetCommand(final OffsetCommand request, final SessionContext ctx) {  
128. //统计计数  
129.         this.statsManager.statsOffset(request.getTopic(), request.getGroup(), 1);  
130. //获取topic对应分区的MessageStore实例  
131.         final MessageStore store = this.storeManager.getMessageStore(request.getTopic(), request.getPartition());  
132. //如果为空，则返回未找到  
133.         if (store == null) {  
134.             return new BooleanCommand(HttpStatus.NotFound, "The topic `" + request.getTopic() + "` in partition `" + request.getPartition() + "` is not exists", request.getOpaque());  
135.         }  
136.         //获取topic对应分区最近可用的offset  
137. final long offset = store.getNearestOffset(request.getOffset());  
138.         return new BooleanCommand(HttpStatus.Success, String.valueOf(offset), request.getOpaque());  
139.     }  
140.   
141. //退出请求业务逻辑处理  
142.     @Override  
143.     public void processQuitCommand(final QuitCommand request, final SessionContext ctx) {  
144.         try {  
145.             if (ctx.getConnection() != null) {  
146.                 //关闭与客户端的连接  
147.                 ctx.getConnection().close(false);  
148.             }  
149.         } catch (final NotifyRemotingException e) {  
150.             // ignore  
151.         }  
152.     }  
153.   
154. //版本查询请求业务逻辑处理  
155. @Override  
156.     public ResponseCommand processVesionCommand(final VersionCommand request, final SessionContext ctx) {  
157. //返回当前Broker版本  
158.         return new BooleanCommand(HttpStatus.Success, BuildProperties.VERSION, request.getOpaque());  
159.     }  
160.   
161. //统计请求查询业务逻辑处理  
162.     @Override  
163.     public ResponseCommand processStatCommand(final StatsCommand request, final SessionContext ctx) {  
164. //判断类型，如果类型以config 开头，则传输整个配置文件  
165.         final String item = request.getItem();  
166.         if ("config".equals(item)) {  
167.             return this.processStatsConfig(request, ctx);  
168.         } else {  
169. //如果是获取统计结果，则从统计模块获取响应结果并返回给客户端  
170.             final String statsInfo = this.statsManager.getStatsInfo(item);  
171.             return new BooleanCommand(HttpStatus.Success, statsInfo, request.getOpaque());  
172.         }  
173.     }  
174.   
175.     //获取配置文件内容，使用zeroCopy将文件内容发送到客户端，构造的响应用BooleanCommand  
176. @SuppressWarnings("resource")  
177.     private ResponseCommand processStatsConfig(final StatsCommand request, final SessionContext ctx) {  
178.         try {  
179.             final FileChannel fc = new FileInputStream(this.metaConfig.getConfigFilePath()).getChannel();  
180.             // result code length opaque\r\n  
181.             IoBuffer buf = IoBuffer.allocate(11 + 3 + ByteUtils.stringSize(fc.size()) + ByteUtils.stringSize(request.getOpaque()));  
182.             ByteUtils.setArguments(buf, MetaEncodeCommand.RESULT_CMD, HttpStatus.Success, fc.size(), request.getOpaque());  
183.             buf.flip();  
184.             ctx.getConnection().transferFrom(buf, null, fc, 0, fc.size(), request.getOpaque(),  
185.                     new SingleRequestCallBackListener() {  
186.                         @Override  
187.                         public void onResponse(ResponseCommand responseCommand, Connection conn) {  
188.                             this.closeChannel();  
189.                         }  
190.   
191.                         @Override  
192.                         public void onException(Exception e) {  
193.                             this.closeChannel();  
194.                         }  
195.   
196.                         private void closeChannel() {  
197.                             try {  
198.                                 fc.close();  
199.                             } catch (IOException e) {  
200.                                 log.error("IOException while stats config", e);  
201.                             }  
202.                         }  
203.   
204.                         @Override  
205.                         public ThreadPoolExecutor getExecutor() {  
206.                             return null;  
207.                         }  
208.                     }, 5000, TimeUnit.MILLISECONDS);  
209.         } catch (FileNotFoundException e) {  
210.             log.error("Config file not found:" + this.metaConfig.getConfigFilePath(), e);  
211.             return new BooleanCommand(HttpStatus.InternalServerError, "Config file not found:" + this.metaConfig.getConfigFilePath(), request.getOpaque());  
212.         } catch (IOException e) {  
213.             log.error("IOException while stats config", e);  
214.             return new BooleanCommand(HttpStatus.InternalServerError, "Read config file error:" + e.getMessage(), request.getOpaque());  
215.         } catch (NotifyRemotingException e) {  
216.             log.error("NotifyRemotingException while stats config", e);  
217.         }  
218.         return null;  
219.     }  

 

如果不使用内容的事务，Broker已经完成了从网络接收数据—>处理请求(存储消息/查询结果等)—>返回结果的流程，Broker最基础的流程已经基本分析完毕。