二分实现及工程使用—Kafka

典型的二分思想：“猜数字”的游戏。大家规定一个范围，一个人在心里想一个这个范围内的具体数字，比如一个 1-100 的自然数，然后另几个人来猜数字；每次猜错，这个人都会提示他们的猜测是大了还是小了，看谁最快猜到数字。

大家的第一反应也都会是从比较中间的位置，比如 50，开始猜起。毕竟如果 50 猜错了，因为要提示是大了还是小了，范围就要么缩小到 1-49，要么缩小到 51-100，这样猜测范围就可以成倍的缩小。

如果每一次我们都猜测可能范围内的中间值，那么即使猜错了也能成倍的缩小范围，这样的策略其实就是二分查找算法。

有了二分查找算法，即使更大的范围内进行游戏，比如在 1-1,000,000 的范围内，我们按照二分的策略，最多也只需要 20 次即可完成任意数字的猜测，极大的减少的猜数字所需的时间

二分查找

比如在刚刚说的猜数字游戏里，我们之所以每次能排除一半的搜索空间，就是因为数字整体是有序排列的，如果某次猜测的数 x，比目标值 target 大，那么当然比 x 更大的数就没有必要猜测了。

百度：

二分查找是一种在有序数组中查找某一特定元素的搜索算法。搜索过程从数组的中间元素开始，如果中间元素正好是要查找的元素，则搜索过程结束；如果某一特定元素大于或者小于中间元素，则在数组大于或小于中间元素的那一半中查找，而且跟开始一样从中间元素开始比较。如果在某一步骤数组为空，则代表找不到。这种搜索算法每一次比较都使搜索范围缩小一半。

当搜索空间里只剩一个可能元素时，也就是最后一次猜测，我们要么猜到了答案，要么就是答案不存在。这样最坏的搜索次数就是最大搜索空间折半多少次可以变成 1。所以二分搜索的时间复杂度就是 O(logn) 了。

下面我们用代码来实现一下：

int guess(int num); // num比答案高返回-1； 否则返回1

class Solution {
public:
int guessNumber(int n) {
int l = 1;
int r = n;

    while(l < r) {
        int mid = l + (r-l)/2; // 每次用左右边界的中点作为猜测值

        if (guess(mid) == 0) return mid; //猜中直接返回

        if (guess(mid) < 0) { // 猜的数大了
            r = mid - 1;
        } else { // 猜的数小了
            l = mid+1;
        }
    }

    return l;
}

};

Kafka中二分的使用

Kafka 的海量消息就是按照写入的时间顺序，依次追加在许多日志文件中。那在某个日志文件中，每条消息自然会距离第一条消息有一个对应的 offset，不过这里的 offset 更像是一个消息的自增 ID，而不是一个消息在文件中的偏移量。

怎么找到日志

Kafka 虽然不允许从尾部以外的地方插入或者修改数据，但我们在 Kafka 中还是很可能需要从某个时间点开始读数据的，这就意味着我们要通过一个 offset，快速查找到某条消息在日志文件的什么位置。这里再强调一下，kafka 中的 offset 实际上是类似于消息自增 ID 的存在，并不是真的在磁盘上的偏移量。

00000000000000000000.log
00000000000000000000.index
00000000000000000000.timeindex
00000000000000000035.log
00000000000000000035.index
00000000000000000035.timeindex

.log 文件就是存储了消息体本身的日志文件；

.index 文件就是用于帮我们快速检索消息在文件中位置的索引文件；

这里还有个.timeindex 后缀的文件，它和 index 其实差不多都是索引文件，只不过在这个文件中关联 position 的变成了时间戳。

在整个 Kafka 中，二分搜索的核心作用就是用于加速索引指定 offset 的消息，所以相应的代码都在 core/src/main/scala/kafka/log/AbstractIndex.scala 中。 indexSlogRangeFor 就是用于检索目标值的函数，其返回值就代表可能范围的上下界，我们会不断的递归搜索，如果最终返回的下界和上界相等，就说明我们找到了目标值。

private def indexSlotRangeFor(idx: ByteBuffer, target: Long, searchEntity: IndexSearchEntity): (Int, Int) = {
  // 检查index是否为空
  if(_entries == 0)
    return (-1, -1)

    // 二分搜索
  def binarySearch(begin: Int, end: Int) : (Int, Int) = {
    var lo = begin
    var hi = end
    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)
  }

  val firstHotEntry = Math.max(0, _entries - 1 - _warmEntries)
  // 查询的目标offset是否在热区
  if(compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {
    return binarySearch(firstHotEntry, _entries - 1)
  }

  // 查询的目标offset是否小于最小的offset
  if(compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0)
    return (-1, 0)

  return binarySearch(0, firstHotEntry)
}