day-3

BufferPool

• 产生原因：ByteBuffer的创建和释放都是比较耗费资源的，为了实现内存的高效利用，产生了他。他会对特定大小的ByteBuffer进行管理

BufferPool的字段

• free:是一个ArrayDeque队列，缓存指定大小的ByteBuffer对象
• ReentrantLock:因为这个BufferPool在多线程中使用，保证其安全
• waiters:这个队列中包含小空间导致阻塞的线程对应的Condition对象
• totalMemory:整个pool的大小
• availableMemory:可用空间大小-totalMemeoy-free中每个ByteBuffer大小

BufferPool的函数

• allocate():从pool中申请ByteBuffer，要是因为空间不足申请不下来的时候，就阻塞调用线程
• deallocate:从pool中释放大小为poolableSize[他是pool队列里面所有ByteBuffer的大小],然后唤醒一个因空间不足而阻塞的线程

RecordAccumulator

RecordAccumulator字段

• batches:类型CopyOnWriteMap,TopicPartition-RecordBatch集合，他是一个队列，每个队员里面都可以找到RecordBatch集合要发送到的目的地TopicPartition
• BatchSize:指定每个RecordBatch底层的ByteBuffer大小
• Compression：压缩类型
• imcomplete:未发送完成的RecordBatch集合
• free:BufferPool对象
• drainIndex:发送RecordBatch集合一次发不完，他记录的就是下次开始发送的位置

kafkaProducer.send()调用的时候，最终内部调用的是RecordAcculator.append()

RecordAcculator.append()详细步骤

• 在batch中找TopicPartition对应的RecordBatch集合，要是没有，就创建，并将他添加到batches
• 对Deque加锁（Deque就是RecordBatch集合）
• 调用tryAppend(),想Deque最后一个RecordBatch追加Record(消息)
• 对Deque解锁
• 要是成功：返回RecordAppendResult[内部封装了ProduceRequestRequest]唤醒sender;
• 追加失败：从BufferPool中申请新的ByteBuffer，回到给Deque加锁继续直到追加成功，要是还是失败，就用ByteBuffer创建RecordBatch, 将Record追加到新建的RecordBatch,将RecordBatch追加打破Deque最后将新建的RecordBatch追加到incomplete集合,然后解锁，返回并唤醒sender;
  ####### 当然，这里唤醒sender也不是说唤醒就能唤醒的，他是调用了kafkaProducer.dosend()函数判断此次向RecordAccumlator中追加消息是否满足消息所在的最后一个RecordBatch满了，或者队列中不指一个RecordBatch

Ready()[在客户端向服务器发送消息前，会调用它来获取符合发送的消息集合的节点],有如下条件

• Deque中有多个RecordBatch或者第一个RecordBatch是否满了【只取第一个进行判断就行】
• 是否超时了
• 是否BufferPool的空间耗尽了
• 是否有线程在等待flush操作完成
• sender线程准备关闭

他的过程如下：首先他会遍历batches集合的每分区，找到分区leader所在的node,要是满足上面的条件就加到readyNodes集合中；过程完后最后函数返回ReadyCheckResult对象，他里面记录了满足的node集合，还有直到步道leader的分区，还有下次调用ready()进行检查的时间间隔

kafkaProducer.send()调用的时候，最终内部调用的是RecordAcculator.append(),接着调用RecordAccumulator.drain()

RecordAccumulator.drain()会将之前的map(TopicPartition,list)构造成map(NodeID,List)返回

• 为什么要这样做呢？—原因：在I/O层，生产者是面向Node发送消息的，但是在sender呢，他是关心消息发送到哪个分区的，因此要进行转换
• 今天也是美好的一天* o*