可关闭的线程抽象类！ 继承自Thread同时还接收一个boolean变量isInterruptible表明是否允许中断。既然是可关闭的，因此一定不是守护线程，而是一个用户线程(不会阻塞JVM关闭)。提供的方法有:

 

1. doWork: 抽象方法。子类必须实现这个方法，从名字来说应该是指定线程要完成的操作。

2. initiateShutdown: 发起关闭请求。首先通过CAS的方式判断是否线程在运行中；如果是的话将表明运行状态的isRunning变量设置为false，同时根据该线程的可中断性(通过isInterruptible)调用Thread.interupt方法中断该线程并返回true。

3. awaitShutdown: 用于发起关闭请求(initiateShutdown)之后调用该方法等待关闭操作完全结束。实现方式是调用CountDownLatch的await方法等待shutdownLacth对象变为0——shutdownLatch是该类维护的一个关闭阀门。在运行线程结束之后程序会将该对象减为0。

4. run方法: 复写了Thread的run方法。该类维护的isRunning类似于一个状态标志，控制着线程何时退出执行操作的循环。每次循环前检测isRunning值，如果一旦发现是false，则退出执行并将shutdownLatch阀门至于关闭状态；如果是true，则调用doWork来执行真正的操作。

 

主要目的是限制某些操作的执行速度，其实主要用于清理日志时限制IO速度。这个类会接收一个给定的期望速率(单位是**/每秒，这里的**其实不重要，可以是字节或个数，主要是限制速率)。构造函数参数如下:
 

1. desiredRatePerSec: 期望速率

2. checkIntervalMs: 检查间隔，单位ms

3. throttleDown: 是否需要往下调节速度，即降低速率

4. metricName: 待调节项名称

5. units: 待调节项单位，默认是字节

6. time: 时间字段

该类还实现了KafkaMetricsGroup trait，你可以认为后者就是构造度量元对象用的(例如通过newMeter)。Throttle类只有一个方法: maybeThrottle。该方法代码写了一大堆，一句一句分析太枯燥，我直接举个例子说吧: 假设我们要限制IO速率，单位是字节/秒，每100毫秒查一次。我们想要限制速率为10字节/毫秒。现在我们在500ms内检测到一共发送了6000字节，那么实际速率是6000/500 = 12字节/毫秒，比期望速率要高，因此我们要限制IO速率，此时怎么办呢？很简单，如果是按照期望速率，应该花费6000/10 = 600ms，比实际多花了100ms，因此程序sleep 100ms把那段多花的时间浪费掉就起到了限制速率的效果。简单来说程序就是这么实现的: )

 

封装了一些与时间相关的常量，另外提供了很多抽象方法供子类实现，比如milliseconds、nanoseconds和sleep。最后真的提供了一个子类实现了Time trait并实现了前面的三个方法。

 

使用jopt-simple工具库验证命令行参数提供的host/port格式是否正确，validatePortOrDie方法对接收到的host: port字符串进行split，过滤掉那些不符合host: port格式的项得到一个Array[String]，如果这个数组的个数与split出来的Array长度不匹配，说明有不合格项，直接打印调用方法并结束程序。

按照这个文件中注释提示的那样，这个文件里面的帮助函数都是最最常用的函数，并不限定为Kafka的逻辑服务。因此如果要在这个文件中添加一个函数，一定要确保这个函数有很完善的文档另外还要保证一定不能只在某个特定的领域使用，应该适用于最常见的场景中。

okay，反正是一个个独立的function，还是老规矩，一个一个说:
 

1. runnable: 接收一个无返回值的函数封装到一个新建的Java Runnable对象中返回。

2. daemonThread: 创建一个守护线程用于后台执行。注意，只是创建，不开启该线程。一共提供了3个方法，目的都是一样

3. newThread: 创建新的线程，参数决定线程名称、执行操作以及是否为守护线程

4. readBytes: 读取一段给定的ByteBuffer以及给定位移到一个字节数组

5. loadProps: 加载一个给定路径下的.properties文件到一个Properties对象中

6. openChannel: 根据可变性打开一个文件通道(file channel)，如果可以修改，以RandomAccessFile方法打开，如果不可修改，以FileInputStream方法打开

7. swallow: 执行给定的操作(以函数参数方式传递), 记录下任何异常但绝不抛出异常，而是吞掉异常

8. equal: 比较两段ByteBuffer——先比较position，然后是remaining，最后是内部的每个字节

9. readString: 将ByteBuffer中的内容读出来生成一个字符串

10. croak: 打印错误消息并关闭JVM

11. rm: 递归地删除文件或子目录

12. registerMBean: 将给定的mbean注册到平台mbean服务器。如果已经注册过会先卸载该mbean再重新注册。该方法不会抛出任何异常，只是简单滴返回false表示注册失败

13. unregisterMBean: 取消mbean的注册。如果没有注册过，就什么都不做

14. readUnsignedInt: 一共有2个方法，一个是从ByteBuffer当前位置读取4个字节的integer，并更新位移；另一个是从给定位置读取4个字节的integer，但并不修改position信息

15. writeUnsignedInt: 也是提供了2个方法: 一个是将一个long型作为一个4个字节的integer写入ByteBuffer，不考虑溢出的问题；第二个方法就是在给定的位置做第一个方法的事情

16. crc32: 计算字节数组的CRC32校验码

17. hashcode: 计算给定参数项的hashCode

18. groupby: 按照给定的函数计算出来的key进行分组，将相同key的值都放入一个List[V]中,整体作为一个HashMap返回。不过貌似这个函数没有被用到: (

19. read: 读取给定ReadableByteChannel的buffer到一个ByteBuffer，如果读取失败抛出EOFException异常，否则返回读取的字节数

20. notNull: 带泛型类型的方法，如果给定值是null抛出异常，否则直接返回该值

21. stackTrace: 获取给定异常类的栈追踪信息(stack trace)，作为字符串返回

22. parseCsvMap: 接收一个逗号分隔的key/value对，类似于key1: value1,key2: value2,... 使用正则表达式解析成一个HashMap并返回。该方法还有个形式，是返回成一个字符串序列，即字符串数组

23. createObject: 通过反射机制为给定类名表示的类创建一个实例

24. nullOrEmpty: 判断给定字符串是否为空。但实现有个问题，如果是多个空格，该方法就会返回false

25.  loopIterator: 其实就是个无限遍历的迭代器，为传入的集合生成一个无限流Stream，并返回它的迭代器

26. readFileAsString: 读取文件内容并作为一个字符串返回

27. abs: 计算某个整数的绝对值。如果是最小整数返回0——与Java的实现不同

28. replaceSuffix: 替换字符串的后缀，如果无法找到要替换的后缀直接抛出异常

29. createFile: 根据给定路径创建文件

30. asString: 将一个Properties对象转换成字符串返回

31. readProps: 从给定的字符串中读取属性并返回一个Properties

32. readInt: 从字节数组中的指定位移处读取一个integer

33. inLock: 在加锁的情况下执行一段函数

34. inReadLock/inWriterLock: 使用给定的锁分别获取一个读锁和写锁，然后执行函数体

35. JSONEscapeString: 将字符串中的某些字符进行转义，比如\b转成\\b

36. duplicates: 返回列表中重复项

 

这个类就是封装了Properties对象，同时维护了一个HashSet表示属性名称集合。具体的方法有：

1. containsKey: 判断是否包含某个key

2. getProperty: 将属性名加入到属性名称集合，然后从props中获取某个属性值之后返回

3. getString: 先检测是否包含这个属性，如果包含则调用getProperty返回属性值

4. getInt: 获取一个integer类型的属性值

5. getIntInRange: 返回一个integer类型的属性值，但该值必须在给定的范围内

6. getShort: 返回一个short类型的属性值

7. getShortInRange: 返回一个short类型的属性值，但该值必须在 给定的范围内

8. getLong: 返回一个long类型的属性值

9. getLongInRange: 返回一个long类型的属性值，但该值必须在给定的范围内

10. getDouble: 返回一个double类型的属性值

11. getBoolean: 返回一个boolean类型的属性值

12. getMap: 从一个属性列表中解析出一个Map[String, String]并返回

13. getCompressionCodec: 从属性列表中读取处codec信息。该方法同时支持解析codec的序号和名称，并返回对应的codec

14. verify: 主要就是验证Properties对象中每个属性是否都在属性名称集合中，即使不在也只是打印一个log而已

 

终于到包里面的最后一个文件了。这个文件代码很长，定义了三个class和两个object。先从简单的开始说吧：

1. ZKConfig class: 这个类定义了zookeeper配置信息，其构造函数接收一个VerifiableProperties对象。Kafka维护的zookeeper配置信息包括：一个CSV格式的zookeeper连接串、最大会话超时时间、连接最大超时时间以及一个zookeeper follower被允许落后于leader的最大时间间隔，默认是2秒

2. ZKGroupDirs class: 名字中的group是指kafka的消费组的，因为Kafka中的consumer都属于一个consumer group。在这个类中定义了3个方法分别返回消费者的根路径，消费者组的路径以及消费者组的id的路径

3. ZKGroupTopicDirs class: 该类继承了ZKGroupDirs类，并提供了2个新的方法用于计算消费者组的位移和拥有者信息。如下图所示：