Kafka原始碼解析（二）---Log分析
阿新 • • 發佈：2020-06-21
上一篇文章講了LogSegment和Log的初始化，這篇來講講Log的主要操作有哪些。
一般來說Log 的常見操作分為 4 大部分。
1. 高水位管理操作
2. 日誌段管理
3. 關鍵位移值管理
4. 讀寫操作
其中關鍵位移值管理主要包含Log Start Offset 和 LEO等。
## 高水位HighWatermark
### 高水位HighWatermark初始化
高水位是通過LogOffsetMetadata類來定義的：
```scala
@volatile private var highWatermarkMetadata: LogOffsetMetadata = LogOffsetMetadata(logStartOffset)
```
這裡傳入的初始值是logStartOffset，表明當首次構建高水位時，它會被賦值成 Log Start Offset 值。
我們再來看看LogOffsetMetadata類：
```scala
case class LogOffsetMetadata(messageOffset: Long,
segmentBaseOffset: Long = Log.UnknownOffset,
relativePositionInSegment: Int = LogOffsetMetadata.UnknownFilePosition) {
// check if this offset is already on an older segment compared with the given offset
def onOlderSegment(that: LogOffsetMetadata): Boolean = {
if (messageOffsetOnly)
throw new KafkaException(s"$this cannot compare its segment info with $that since it only has message offset info")
this.segmentBaseOffset < that.segmentBaseOffset
}
...
}
```
LogOffsetMetadata有三個初始值：
messageOffset表示訊息位移值；
segmentBaseOffset儲存訊息位移值所在日誌段的起始位移，用來判斷兩條訊息是否處於同一個日誌段的；
relativePositionSegment儲存訊息位移值所在日誌段的物理磁碟位置；
上面的onOlderSegment表明，要比較哪個日誌段更老，只需要比較segmentBaseOffset的大小就可以了。
### 高水位HighWatermark設值與更新
```scala
private def updateHighWatermarkMetadata(newHighWatermark: LogOffsetMetadata): Unit = {
//高水位的值不可能小於零
if (newHighWatermark.messageOffset < 0)
throw new IllegalArgumentException("High watermark offset should be non-negative")
lock synchronized {// 保護Log物件修改的Monitor鎖
highWatermarkMetadata = newHighWatermark// 賦值新的高水位值
//事務相關，暫時忽略
producerStateManager.onHighWatermarkUpdated(newHighWatermark.messageOffset)
//事務相關，暫時忽略
maybeIncrementFirstUnstableOffset()
}
trace(s"Setting high watermark $newHighWatermark")
}
```
設定高水位的值是很簡單的，首先校驗高水位的值是否大於零，然後通過直接加鎖之後更新高水位的值。
更新更新高水位值的方法有兩個：updateHighWatermark 和 maybeIncrementHighWatermark，我們分別分析。
**updateHighWatermark**
```scala
def updateHighWatermark(hw: Long): Long = {
//傳入的高水位的值如果小於logStartOffset，設定為logStartOffset
val newHighWatermark = if (hw < logStartOffset)
logStartOffset
// 傳入的高水位的值如果大於LEO，那麼設定為LEO
else if (hw > logEndOffset)
logEndOffset
else
hw
//將newHighWatermark封裝成一個LogOffsetMetadata然後更新高水位的值
updateHighWatermarkMetadata(LogOffsetMetadata(newHighWatermark))
//返回新的高水位的值
newHighWatermark
}
```
這個方法邏輯也很簡潔，因為高水位的值是不可能大於LEO，也不可能小於logStartOffset，所以需要對傳入的hw校驗然後設定成正確的值，然後呼叫上面的設定高水位的方法設值。
**maybeIncrementHighWatermark**
```scala
/**
* Update the high watermark to a new value if and only if it is larger than the old value. It is
* an error to update to a value which is larger than the log end offset.
*
* This method is intended to be used by the leader to update the high watermark after follower
* fetch offsets have been updated.
*
* @return the old high watermark, if updated by the new value
*/
// 當新的高水位的值大於舊的高水位的值時才做更新，如果新的高水位的值大於LEO，會報錯
// 這個方法是leader在確認Follower已經拉取了日誌之後才做更新
def maybeIncrementHighWatermark(newHighWatermark: LogOffsetMetadata): Option[LogOffsetMetadata] = {
//如果新的高水位的值大於LEO，會報錯
if (newHighWatermark.messageOffset >