class ByteBufferMessageSet(val buffer: ByteBuffer) extends MessageSet with Logging { //这里作为入口传入ByteBuffer类型的buffer

  private var shallowValidByteCount = -1  //设定了全局的验证字节统计变量.

  //这里是几个不同的构造函数.但是都是以最初的构造函数为基础的.这里调用了BBMS对象的create方法创建新buffer.

  def this(compressionCodec: CompressionCodec, messages: Message*) {    //参数是压缩标志和message

    this(ByteBufferMessageSet.create(new AtomicLong(0), compressionCodec, messages:_*))

  }

  //同上.只是参数不同而已.

  def this(compressionCodec: CompressionCodec, offsetCounter: AtomicLong, messages: Message*) {

    this(ByteBufferMessageSet.create(offsetCounter, compressionCodec, messages:_*))

  }

 //最简单的消息构造函数.

  def this(messages: Message*) {

    this(NoCompressionCodec, new AtomicLong(0), messages: _*)

  }

object ByteBufferMessageSet {

  //创建函数的参数是offset,压缩,message.返回的是bytebuffer类型.

  private def create(offsetCounter: AtomicLong, compressionCodec: CompressionCodec, messages: Message*): ByteBuffer = {

    if(messages.size == 0) {  //消息为空的话就返回一个空的buffer

      MessageSet.Empty.buffer

    } else if(compressionCodec == NoCompressionCodec) {  //没有压缩的话.就分配一个新的messages大小的buffer.

      val buffer = ByteBuffer.allocate(MessageSet.messageSetSize(messages))

      for(message <- messages)

        writeMessage(buffer, message, offsetCounter.getAndIncrement) //在这里将消息内容写入buffer中.

      buffer.rewind() //将指针设置到开始位置.

      buffer //返回buffer

    } else { //这里则是启用压缩的处理动作了.

      val byteArrayStream = new ByteArrayOutputStream(MessageSet.messageSetSize(messages)) //获取一个messages大小的字节流对象

      val output = new DataOutputStream(CompressionFactory(compressionCodec, byteArrayStream)) //创建写入对象.

      var offset = -1L

      try {

        for(message <- messages) { //将消息压缩写入字节流对象中去.

          offset = offsetCounter.getAndIncrement

          output.writeLong(offset)

          output.writeInt(message.size)

          output.write(message.buffer.array, message.buffer.arrayOffset, message.buffer.limit)

        }

      } finally {

        output.close()

      }

      val bytes = byteArrayStream.toByteArray //转成字节

      val message = new Message(bytes, compressionCodec) //创建一个压缩的消息对象.

      val buffer = ByteBuffer.allocate(message.size + MessageSet.LogOverhead) /分配一个buffer

      writeMessage(buffer, message, offset) //将压缩过的message写入到buffer.

      buffer.rewind()

      buffer

    }

  }

  def decompress(message: Message): ByteBufferMessageSet = {

    val outputStream: ByteArrayOutputStream = new ByteArrayOutputStream

    val inputStream: InputStream = new ByteBufferBackedInputStream(message.payload)

    val intermediateBuffer = new Array[Byte](1024)

    val compressed = CompressionFactory(message.compressionCodec, inputStream)

    try {

      Stream.continually(compressed.read(intermediateBuffer)).takeWhile(_ > 0).foreach { dataRead =>

        outputStream.write(intermediateBuffer, 0, dataRead)

      }

    } finally {

      compressed.close()

    }

    val outputBuffer = ByteBuffer.allocate(outputStream.size)

    outputBuffer.put(outputStream.toByteArray)

    outputBuffer.rewind

    new ByteBufferMessageSet(outputBuffer)

  }

    //将message写入buffer

  private[kafka] def writeMessage(buffer: ByteBuffer, message: Message, offset: Long) {

    buffer.putLong(offset) //先写入offset 之后是size,最后是消息本身.

    buffer.putInt(message.size)

    buffer.put(message.buffer)

    message.buffer.rewind()

  }

}

override def iterator: Iterator[MessageAndOffset] = internalIterator()

  /** iterator over compressed messages without decompressing */

  def shallowIterator: Iterator[MessageAndOffset] = internalIterator(true)

  /** When flag isShallow is set to be true, we do a shallow iteration: just traverse the first level of messages. **/

  private def internalIterator(isShallow: Boolean = false): Iterator[MessageAndOffset] = { //可以看见返回的是一个迭代对象,对象类想是messageandoffset

    new IteratorTemplate[MessageAndOffset] {  //这是一个抽象类.在这个地方返回了一个匿名类的实例.

      var topIter = buffer.slice()  //复制一份buffer

      var innerIter: Iterator[MessageAndOffset] = null

      def innerDone():Boolean = (innerIter == null || !innerIter.hasNext)

      def makeNextOuter: MessageAndOffset = { //这里主要把没个message映射成messageandoffset

        // if there isn't at least an offset and size, we are done

        if (topIter.remaining < 12)

          return allDone()

        val offset = topIter.getLong()

        val size = topIter.getInt()

        if(size < Message.MinHeaderSize)

          throw new InvalidMessageException("Message found with corrupt size (" + size + ")")

        // we have an incomplete message

        if(topIter.remaining < size)

          return allDone()

          //上面这些主要是检查啥的.不细说了.主要映射在下面完成.

        // read the current message and check correctness

        val message = topIter.slice()

        message.limit(size) //通过size截取第一个message

        topIter.position(topIter.position + size) //将指针往前移.

        val newMessage = new Message(message) //构建新message

        if(isShallow) {

          new MessageAndOffset(newMessage, offset) //做映射返回.

        } else { //剩下是判断是否有压缩的情况.

          newMessage.compressionCodec match {

            case NoCompressionCodec =>

              innerIter = null

              new MessageAndOffset(newMessage, offset)

            case _ =>

              innerIter = ByteBufferMessageSet.decompress(newMessage).internalIterator() //有压缩用对象里的方法处理.

              if(!innerIter.hasNext)

                innerIter = null

              makeNext()

          }

        }

      }

     //在这个函数里调用映射函数.最后这个函数有next函数调用.

      override def makeNext(): MessageAndOffset = {

        if(isShallow){

          makeNextOuter

        } else {

          if(innerDone())

            makeNextOuter

          else

            innerIter.next

        }

      }

    }

  }

abstract class IteratorTemplate[T] extends Iterator[T] with java.util.Iterator[T] {

  private var state: State = NOT_READY

  private var nextItem = null.asInstanceOf[T]

  def next(): T = {

    if(!hasNext())

      throw new NoSuchElementException()

    state = NOT_READY

    if(nextItem == null)

      throw new IllegalStateException("Expected item but none found.")

    nextItem

  }

  def peek(): T = {

    if(!hasNext())

      throw new NoSuchElementException()

    nextItem

  }

  def hasNext(): Boolean = {

    if(state == FAILED)

      throw new IllegalStateException("Iterator is in failed state")

    state match {

      case DONE => false

      case READY => true

      case _ => maybeComputeNext()

    }

  }

  protected def makeNext(): T

  def maybeComputeNext(): Boolean = {

    state = FAILED

    nextItem = makeNext()

    if(state == DONE) {

      false

    } else {

      state = READY

      true

    }

  }

  protected def allDone(): T = {

    state = DONE

    null.asInstanceOf[T]

  }

  def remove =

    throw new UnsupportedOperationException("Removal not supported")

  protected def resetState() {

    state = NOT_READY

  }

}