这里主要更新一下kafka 0.10.0版本的message消息格式的变化。
message 的格式在0.10.0的版本里发生了一些变化（相对于0.8.2.1的版本）这里把0.10.0的message的代码分析一下。
message的格式定义在源代码目录message目录里的Message.scala的源代码目录里。下面看一下主要代码。
object Message {
/**
* The current offset and size for all the fixed-length fields
*/
val CrcOffset = 0
val CrcLength = 4
val MagicOffset = CrcOffset + CrcLength
val MagicLength = 1
val AttributesOffset = MagicOffset + MagicLength
val AttributesLength = 1
// Only message format version 1 has the timestamp field.
val TimestampOffset = AttributesOffset + AttributesLength //可以看见最明显的变化在这里。在以前的message格式的基础上head部分多了一个字段。就是时间戳的字段。
val TimestampLength = 8 //时间戳的字段长度是8个字节。 唯一的变化在这里。
val KeySizeOffset_V0 = AttributesOffset + AttributesLength //同时支持了消息的两个版本。v0和v1，v0跟以前一样。没有时间戳的字段。
val KeySizeOffset_V1 = TimestampOffset + TimestampLength //v1多了时间戳的字段。
val KeySizeLength = 4
val KeyOffset_V0 = KeySizeOffset_V0 + KeySizeLength
val KeyOffset_V1 = KeySizeOffset_V1 + KeySizeLength
val ValueSizeLength = 4
//下面的一些定义都是针对消息的相关定义。跟之前也有些变化但是变化不大。
private val MessageHeaderSizeMap = Map (
(0: Byte) -> (CrcLength + MagicLength + AttributesLength + KeySizeLength + ValueSizeLength),
(1: Byte) -> (CrcLength + MagicLength + AttributesLength + TimestampLength + KeySizeLength + ValueSizeLength))
/**
* The amount of overhead bytes in a message
* This value is only used to check if the message size is valid or not. So the minimum possible message bytes is
* used here, which comes from a message in message format V0 with empty key and value.
*/
val MinMessageOverhead = KeyOffset_V0 + ValueSizeLength
/**
* The "magic" value
* When magic value is 0, the message uses absolute offset and does not have a timestamp field.
* When magic value is 1, the message uses relative offset and has a timestamp field.
*/
val MagicValue_V0: Byte = 0
val MagicValue_V1: Byte = 1
val CurrentMagicValue: Byte = 1
/**
* Specifies the mask for the compression code. 3 bits to hold the compression codec.
* 0 is reserved to indicate no compression
*/
val CompressionCodeMask: Int = 0x07
/**
* Specifies the mask for timestamp type. 1 bit at the 4th least significant bit.
* 0 for CreateTime, 1 for LogAppendTime
*/
val TimestampTypeMask: Byte = 0x08
val TimestampTypeAttributeBitOffset: Int = 3
/**
* Compression code for uncompressed messages
*/
val NoCompression: Int = 0
/**
* To indicate timestamp is not defined so "magic" value 0 will be used.
*/
val NoTimestamp: Long = -1
/**
* Give the header size difference between different message versions.
*/
def headerSizeDiff(fromMagicValue: Byte, toMagicValue: Byte) : Int =
MessageHeaderSizeMap(toMagicValue) - MessageHeaderSizeMap(fromMagicValue)
}
这里就简单记录一下v0.10.0的message格式上的变化。对于Message.scala源代码里的方法，变量等就不再做详细说明了。