You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

367 lines
23 KiB
Markdown

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

# 04 | 索引改进的二分查找算法在Kafka索引的应用
你好我是胡夕。今天我来带你学习一下Kafka源码中的索引对象以及改进版二分查找算法Binary Search Algorithm在索引中的应用。
## 为什么要阅读索引源码?
坦率地说你在Kafka中直接接触索引或索引文件的场景可能不是很多。索引是一个很神秘的组件Kafka官方文档也没有怎么提过它。你可能会说既然这样我还有必要读索引对象的源码吗其实是非常有必要的我给你分享一个真实的例子。
有一次我用Kafka的DumpLogSegments类去查看底层日志文件和索引文件的内容时发现了一个奇怪的现象——查看日志文件的内容不需要sudo权限而查看索引文件的内容必须要有sudo权限如下所示
```
$ sudo ./kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index
Dumping 00000000000000000000.index
offset: 0 position: 0
$ ./kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000000000000.index
Dumping 00000000000000000000.index
Exception in thread "main" java.io.FileNotFoundException: 00000000000000000000.index (Permission denied)
......
```
看了索引源码之后我才知道原来Kafka读取索引文件时使用的打开方式是rw。实际上读取文件不需要w权限只要r权限就行了。这显然是Kafka的一个Bug。你看通过阅读源码我找到了问题的根本原因还顺便修复了Kafka的一个问题[KAFKA-5104](https://issues.apache.org/jira/browse/KAFKA-5104))。
除了能帮我们解决实际问题之外,索引这个组件的源码还有一个亮点,那就是**它应用了耳熟能详的二分查找算法来快速定位索引项**。关于算法,我一直觉得很遗憾的是,**我们平时太注重算法本身,却忽略了它们在实际场景中的应用**。
比如说我们学习了太多的排序算法但是对于普通的应用开发人员来说亲自使用这些算法完成编程任务的机会实在太少了。说起数组排序你可能只记得调用Collections.sort方法了但它底层应用了什么排序算法其实并不清楚。
难得的是Kafka的索引组件中应用了二分查找算法而且社区还针对Kafka自身的特点对其进行了改良。这难道不值得我们好好学上一学吗话不多说现在我们就开始学习吧。
## 索引类图及源文件组织架构
在Kafka源码中跟索引相关的源码文件有5个它们都位于core包的/src/main/scala/kafka/log路径下。我们一一来看下。
* AbstractIndex.scala它定义了最顶层的抽象类这个类封装了所有索引类型的公共操作。
* LazyIndex.scala它定义了AbstractIndex上的一个包装类实现索引项延迟加载。这个类主要是为了提高性能。
* OffsetIndex.scala定义位移索引保存“<位移值,文件磁盘物理位置>”对。
* TimeIndex.scala定义时间戳索引保存“<时间戳,位移值>”对。
* TransactionIndex.scala定义事务索引为已中止事务Aborted Transcation保存重要的元数据信息。只有启用Kafka事务后这个索引才有可能出现。
这些类的关系如下图所示:
![](https://static001.geekbang.org/resource/image/34/e8/347480a2d1ae659d0ecf590b01d091e8.jpg)
其中OffsetIndex、TimeIndex和TransactionIndex都继承了AbstractIndex类而上层的LazyIndex仅仅是包装了一个AbstractIndex的实现类用于延迟加载。就像我之前说的LazyIndex的作用是为了提升性能并没有什么功能上的改进。
所以今天我先和你讲一讲AbstractIndex这个抽象父类的代码。下节课我再重点和你分享具体的索引实现类。
## AbstractIndex代码结构
我们先来看下AbstractIndex的类定义
```
abstract class AbstractIndex(@volatile var file: File, val baseOffset: Long, val maxIndexSize: Int = -1, val writable: Boolean) extends Closeable {
......
}
```
AbstractIndex定义了4个属性字段。由于是一个抽象基类它的所有子类自动地继承了这4个字段。也就是说Kafka所有类型的索引对象都定义了这些属性。我先给你解释下这些属性的含义。
1. **索引文件**file。每个索引对象在磁盘上都对应了一个索引文件。你可能注意到了这个字段是var型说明它是可以被修改的。难道索引对象还能动态更换底层的索引文件吗是的。自1.1.0版本之后Kafka允许迁移底层的日志路径所以索引文件自然要是可以更换的。
2. **起始位移值**baseOffset。索引对象对应日志段对象的起始位移值。举个例子如果你查看Kafka日志路径的话就会发现日志文件和索引文件都是成组出现的。比如说如果日志文件是00000000000000000123.log正常情况下一定还有一组索引文件00000000000000000123.index、00000000000000000123.timeindex等。这里的“123”就是这组文件的起始位移值也就是baseOffset值。
3. **索引文件最大字节数**maxIndexSize。它控制索引文件的最大长度。Kafka源码传入该参数的值是Broker端参数segment.index.bytes的值即10MB。这就是在默认情况下所有Kafka索引文件大小都是10MB的原因。
4. **索引文件打开方式**writable。“True”表示以“读写”方式打开“False”表示以“只读”方式打开。如果我没记错的话这个参数应该是我加上去的就是为了修复我刚刚提到的那个Bug。
AbstractIndex是抽象的索引对象类。可以说它是承载索引项的容器而每个继承它的子类负责定义具体的索引项结构。比如OffsetIndex的索引项是<位移值,物理磁盘位置>对TimeIndex的索引项是<时间戳,位移值>对。基于这样的设计理念AbstractIndex类中定义了一个抽象方法entrySize来表示不同索引项的大小如下所示
```
protected def entrySize: Int
```
子类实现该方法时需要给定自己索引项的大小对于OffsetIndex而言该值就是8对于TimeIndex而言该值是12如下所示
```
// OffsetIndex
override def entrySize = 8
// TimeIndex
override def entrySize = 12
```
说到这儿你肯定会问为什么是8和12呢我来解释一下。
在OffsetIndex中位移值用4个字节来表示物理磁盘位置也用4个字节来表示所以总共是8个字节。你可能会说位移值不是长整型吗应该是8个字节才对啊。
还记得AbstractIndex已经保存了baseOffset了吗这里的位移值实际上是相对于baseOffset的相对位移值即真实位移值减去baseOffset的值。下节课我会给你重点讲一下它这里你只需要知道**使用相对位移值能够有效地节省磁盘空间**就行了。而Broker端参数log.segment.bytes是整型这说明Kafka中每个日志段文件的大小不会超过2^32即4GB这就说明同一个日志段文件上的位移值减去baseOffset的差值一定在整数范围内。因此源码只需要4个字节保存就行了。
同理TimeIndex中的时间戳类型是长整型占用8个字节位移依然使用相对位移值占用4个字节因此总共需要12个字节。
如果有人问你Kafka中的索引底层的实现原理是什么你可以大声地告诉他**内存映射文件即Java中的MappedByteBuffer**。
使用内存映射文件的主要优势在于它有很高的I/O性能特别是对于索引这样的小文件来说由于文件内存被直接映射到一段虚拟内存上访问内存映射文件的速度要快于普通的读写文件速度。
另外在很多操作系统中比如Linux这段映射的内存区域实际上就是内核的页缓存Page Cache。这就意味着里面的数据不需要重复拷贝到用户态空间避免了很多不必要的时间、空间消耗。
在AbstractIndex中这个MappedByteBuffer就是名为mmap的变量。接下来我用注释的方式带你深入了解下这个mmap的主要流程。
```
@volatile
protected var mmap: MappedByteBuffer = {
// 第1步创建索引文件
val newlyCreated = file.createNewFile()
// 第2步以writable指定的方式读写方式或只读方式打开索引文件
val raf = if (writable) new RandomAccessFile(file, "rw") else new RandomAccessFile(file, "r")
try {
if(newlyCreated) {
if(maxIndexSize < entrySize) // 预设的索引文件大小不能太小,如果连一个索引项都保存不了,直接抛出异常
throw new IllegalArgumentException("Invalid max index size: " + maxIndexSize)
// 第3步设置索引文件长度roundDownToExactMultiple计算的是不超过maxIndexSize的最大整数倍entrySize
// 比如maxIndexSize=1234567entrySize=8那么调整后的文件长度为1234560
raf.setLength(roundDownToExactMultiple(maxIndexSize, entrySize))
}
// 第4步更新索引长度字段_length
_length = raf.length()
// 第5步创建MappedByteBuffer对象
val idx = {
if (writable)
raf.getChannel.map(FileChannel.MapMode.READ_WRITE, 0, _length)
else
raf.getChannel.map(FileChannel.MapMode.READ_ONLY, 0, _length)
}
/* set the position in the index for the next entry */
// 第6步如果是新创建的索引文件将MappedByteBuffer对象的当前位置置成0
// 如果索引文件已存在将MappedByteBuffer对象的当前位置设置成最后一个索引项所在的位置
if(newlyCreated)
idx.position(0)
else
idx.position(roundDownToExactMultiple(idx.limit(), entrySize))
// 第7步返回创建的MappedByteBuffer对象
idx
} finally {
CoreUtils.swallow(raf.close(), AbstractIndex) // 关闭打开索引文件句柄
}
}
```
这些代码最主要的作用就是创建mmap对象。要知道AbstractIndex其他大部分的操作都是和mmap相关。
我举两个简单的小例子。
例1如果我们要计算索引对象中当前有多少个索引项那么只需要执行下列计算即可
```
protected var _entries: Int = mmap.position() / entrySize
```
例2如果我们要计算索引文件最多能容纳多少个索引项只要定义下面的变量就行了
```
private[this] var _maxEntries: Int = mmap.limit() / entrySize
```
再进一步,有了这两个变量,我们就能够很容易地编写一个方法,来判断当前索引文件是否已经写满:
```
def isFull: Boolean = _entries >= _maxEntries
```
总之,**AbstractIndex中最重要的就是这个mmap变量了**。事实上AbstractIndex继承类实现添加索引项的主要逻辑也就是**向mmap中添加对应的字段**。
## 写入索引项
下面这段代码是OffsetIndex的append方法用于向索引文件中写入新索引项。
```
def append(offset: Long, position: Int): Unit = {
inLock(lock) {
// 第1步判断索引文件未写满
require(!isFull, "Attempt to append to a full index (size = " + _entries + ").")
// 第2步必须满足以下条件之一才允许写入索引项
// 条件1当前索引文件为空
// 条件2要写入的位移大于当前所有已写入的索引项的位移——Kafka规定索引项中的位移值必须是单调增加的
if (_entries == 0 || offset > _lastOffset) {
trace(s"Adding index entry $offset => $position to ${file.getAbsolutePath}")
mmap.putInt(relativeOffset(offset)) // 第3步A向mmap中写入相对位移值
mmap.putInt(position) // 第3步B向mmap中写入物理位置信息
// 第4步更新其他元数据统计信息如当前索引项计数器_entries和当前索引项最新位移值_lastOffset
_entries += 1
_lastOffset = offset
// 第5步执行校验。写入的索引项格式必须符合要求即索引项个数*单个索引项占用字节数匹配当前文件物理大小,否则说明文件已损坏
require(_entries * entrySize == mmap.position(), entries + " entries but file position in index is " + mmap.position() + ".")
} else {
// 如果第2步中两个条件都不满足不能执行写入索引项操作抛出异常
throw new InvalidOffsetException(s"Attempt to append an offset ($offset) to position $entries no larger than" +
s" the last offset appended (${_lastOffset}) to ${file.getAbsolutePath}.")
}
}
}
```
我使用一张图来总结下append方法的执行流程希望可以帮你更快速地掌握它的实现
![](https://static001.geekbang.org/resource/image/23/ea/236f0731ff2799f0902ff7293cf6ddea.jpg)
## 查找索引项
索引项的写入逻辑并不复杂难点在于如何查找索引项。AbstractIndex定义了抽象方法**parseEntry**用于查找给定的索引项,如下所示:
```
protected def parseEntry(buffer: ByteBuffer, n: Int): IndexEntry
```
这里的“n”表示要查找给定ByteBuffer中保存的第n个索引项在Kafka中也称第n个槽。IndexEntry是源码定义的一个接口里面有两个方法indexKey和indexValue分别返回不同类型索引的<KeyValue>对。
OffsetIndex实现parseEntry的逻辑如下
```
override protected def parseEntry(buffer: ByteBuffer, n: Int): OffsetPosition = {
OffsetPosition(baseOffset + relativeOffset(buffer, n), physical(buffer, n))
}
```
OffsetPosition是实现IndexEntry的实现类Key就是之前说的位移值而Value就是物理磁盘位置值。所以这里你能看到代码调用了relativeOffset(buffer, n) + baseOffset计算出绝对位移值之后调用physical(buffer, n)计算物理磁盘位置,最后将它们封装到一起作为一个独立的索引项返回。
**我建议你去看下relativeOffset和physical方法的实现看看它们是如何计算相对位移值和物理磁盘位置信息的**。
有了parseEntry方法我们就能够根据给定的n来查找索引项了。但是这里还有个问题需要解决那就是我们如何确定要找的索引项在第n个槽中呢其实本质上这是一个算法问题也就是如何从一组已排序的数中快速定位符合条件的那个数。
## 二分查找算法
到目前为止从已排序数组中寻找某个数字最快速的算法就是二分查找了它能做到O(lgN)的时间复杂度。Kafka的索引组件就应用了二分查找算法。
我先给出原版的实现算法代码。
```
private def indexSlotRangeFor(idx: ByteBuffer, target: Long, searchEntity: IndexSearchEntity): (Int, Int) = {
// 第1步如果当前索引为空直接返回<-1,-1>对
if(_entries == 0)
return (-1, -1)
// 第2步要查找的位移值不能小于当前最小位移值
if(compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0)
return (-1, 0)
// binary search for the entry
// 第3步执行二分查找算法
var lo = 0
var hi = _entries - 1
while(lo < hi) {
val mid = ceil(hi/2.0 + lo/2.0).toInt
val found = parseEntry(idx, mid)
val compareResult = compareIndexEntry(found, target, searchEntity)
if(compareResult > 0)
hi = mid - 1
else if(compareResult < 0)
lo = mid
else
return (mid, mid)
}
(lo, if (lo == _entries - 1) -1 else lo + 1)
```
这段代码的核心是第3步的二分查找算法。熟悉Binary Search的话你对这段代码一定不会感到陌生。
讲到这里似乎一切很完美Kafka索引应用二分查找算法快速定位待查找索引项位置之后调用parseEntry来读取索引项。不过这真的就是无懈可击的解决方案了吗
## 改进版二分查找算法
显然不是我前面说过了大多数操作系统使用页缓存来实现内存映射而目前几乎所有的操作系统都使用LRULeast Recently Used或类似于LRU的机制来管理页缓存。
Kafka写入索引文件的方式是在文件末尾追加写入而几乎所有的索引查询都集中在索引的尾部。这么来看的话LRU机制是非常适合Kafka的索引访问场景的。
这里有个问题是当Kafka在查询索引的时候原版的二分查找算法并没有考虑到缓存的问题因此很可能会导致一些不必要的缺页中断Page Fault。此时Kafka线程会被阻塞等待对应的索引项从物理磁盘中读出并放入到页缓存中。
下面我举个例子来说明一下这个情况。假设Kafka的某个索引占用了操作系统页缓存13个页Page如果待查找的位移值位于最后一个页上也就是Page 12那么标准的二分查找算法会依次读取页号0、6、9、11和12具体的推演流程如下所示
![](https://static001.geekbang.org/resource/image/e4/85/e4fc56a301b13afae4dda303e5366085.jpg)
通常来说一个页上保存了成百上千的索引项数据。随着索引文件不断被写入Page #12不断地被填充新的索引项。如果此时索引查询方都来自ISR副本或Lag很小的消费者那么这些查询大多集中在对Page #12的查询因此Page #0、6、9、11、12一定经常性地被源码访问。也就是说这些页一定保存在页缓存上。后面当新的索引项填满了Page #12页缓存就会申请一个新的Page来保存索引项即Page #13
现在最新索引项保存在Page #13中。如果要查找最新索引项原版二分查找算法将会依次访问Page #0、7、10、12和13。此时问题来了Page 7和10已经很久没有被访问过了它们大概率不在页缓存中因此一旦索引开始征用Page #13就会发生Page Fault等待那些冷页数据从磁盘中加载到页缓存。根据国外用户的测试这种加载过程可能长达1秒。
显然这是一个普遍的问题即每当索引文件占用Page数发生变化时就会强行变更二分查找的搜索路径从而出现不在页缓存的冷数据必须要加载到页缓存的情形而这种加载过程是非常耗时的。
基于这个问题社区提出了改进版的二分查找策略也就是缓存友好的搜索算法。总体的思路是代码将所有索引项分成两个部分热区Warm Area和冷区Cold Area然后分别在这两个区域内执行二分查找算法如下图所示
![](https://static001.geekbang.org/resource/image/e1/8e/e135f0a6b3fb43ead51db4ddbad1638e.jpg)
乍一看,该算法并没有什么高大上的改进,仅仅是把搜寻区域分成了冷、热两个区域,然后有条件地在不同区域执行普通的二分查找算法罢了。实际上,这个改进版算法提供了一个重要的保证:**它能保证那些经常需要被访问的Page组合是固定的**。
想想刚才的例子同样是查询最热的那部分数据一旦索引占用了更多的Page要遍历的Page组合就会发生变化。这是导致性能下降的主要原因。
这个改进版算法的最大好处在于,**查询最热那部分数据所遍历的Page永远是固定的因此大概率在页缓存中从而避免无意义的Page Fault**。
下面我们来看实际的代码。我用注释的方式解释了改进版算法的实现逻辑。一旦你了解了冷区热区的分割原理,剩下的就不难了。
```
private def indexSlotRangeFor(idx: ByteBuffer, target: Long, searchEntity: IndexSearchEntity): (Int, Int) = {
// 第1步如果索引为空直接返回<-1,-1>对
if(_entries == 0)
return (-1, -1)
// 封装原版的二分查找算法
def binarySearch(begin: Int, end: Int) : (Int, Int) = {
// binary search for the entry
var lo = begin
var hi = end
while(lo < hi) {
val mid = (lo + hi + 1) >>> 1
val found = parseEntry(idx, mid)
val compareResult = compareIndexEntry(found, target, searchEntity)
if(compareResult > 0)
hi = mid - 1
else if(compareResult < 0)
lo = mid
else
return (mid, mid)
}
(lo, if (lo == _entries - 1) -1 else lo + 1)
}
// 第3步确认热区首个索引项位于哪个槽。_warmEntries就是所谓的分割线目前固定为8192字节处
// 如果是OffsetIndex_warmEntries = 8192 / 8 = 1024即第1024个槽
// 如果是TimeIndex_warmEntries = 8192 / 12 = 682即第682个槽
val firstHotEntry = Math.max(0, _entries - 1 - _warmEntries)
// 第4步判断target位移值在热区还是冷区
if(compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {
return binarySearch(firstHotEntry, _entries - 1) // 如果在热区,搜索热区
}
// 第5步确保target位移值不能小于当前最小位移值
if(compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0)
return (-1, 0)
// 第6步如果在冷区搜索冷区
binarySearch(0, firstHotEntry)
```
## 总结
今天我带你详细学习了Kafka中的索引机制以及社区如何应用二分查找算法实现快速定位索引项。有两个重点你一定要记住。
1. AbstractIndex这是Kafka所有类型索引的抽象父类里面的mmap变量是实现索引机制的核心你一定要掌握它。
2. 改进版二分查找算法社区在标准原版的基础上对二分查找算法根据实际访问场景做了定制化的改进。你需要特别关注改进版在提升缓存性能方面做了哪些努力。改进版能够有效地提升页缓存的使用率从而在整体上降低物理I/O缓解系统负载瓶颈。你最好能够从索引这个维度去思考社区在这方面所做的工作。
![](https://static001.geekbang.org/resource/image/31/22/31e35198818b0bbc05ef333b5897b022.jpg)
实际上无论是AbstractIndex还是它使用的二分查找算法它们都属于Kafka索引共性的东西即所有Kafka索引都具备这些特点或特性。那么你是否想了解不同类型索引之间的区别呢比如位移索引和时间戳索引有哪些异同之处。这些问题的答案我会在下节课揭晓你千万不要错过。
## 课后讨论
目前冷区和热区的分割线设定在8192字节处请结合源码注释以及你自己的理解讲一讲为什么要设置成8192
欢迎你在留言区畅所欲言,跟我交流讨论,也欢迎你把今天的内容分享给你的朋友。