温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

怎样深度剖析Kafka Producer的缓冲池机制

发布时间:2021-12-15 09:25:07 来源:亿速云 阅读:120 作者:柒染 栏目:大数据

这期内容当中小编将会给大家带来有关怎样深度剖析Kafka Producer的缓冲池机制,文章内容丰富且以专业的角度为大家分析和叙述,阅读完这篇文章希望大家可以有所收获。

深度剖析 Kafka Producer 的缓冲池机制

在新版的 Kafka Producer 中,设计了一个消息缓冲池,在创建 Producer 时会默认创建一个大小为 32M 的缓冲池,也可以通过 buffer.memory 参数指定缓冲池的大小,同时缓冲池被切分成多个内存块,内存块的大小就是我们创建 Producer 时传的 batch.size 大小,默认大小 16384,而每个 Batch 都会包含一个 batch.size 大小的内存块,消息就是存放在内存块当中。整个缓冲池的结构如下图所示:

怎样深度剖析Kafka Producer的缓冲池机制

客户端将消息追加到对应主题分区的某个 Batch 中,如果 Batch 已经满了,则会新建一个 Batch,同时向缓冲池(RecordAccumulator)申请一块大小为 batch.size 的内存块用于存储消息。

当 Batch 的消息被发到了 Broker 后,Kafka Producer 就会移除该 Batch,既然 Batch 持有某个内存块,那必然就会涉及到 GC 问题。

频繁的申请内存,用完后就丢弃,必然导致频繁的 GC,造成严重的性能问题。那么,Kafka 是怎么做到避免频繁 GC 的呢?

前面说过了,缓冲池在设计逻辑上面被切分成一个个大小相等的内存块,当消息发送完毕,归还给缓冲池不就可以避免被回收了吗?

缓冲池的内存持有类是 BufferPool,我们先来看下 BufferPool 都有哪些成员:

public class BufferPool {  // 总的内存大小  private final long totalMemory;  // 每个内存块大小,即 batch.size  private final int poolableSize;  // 申请、归还内存的方法的同步锁  private final ReentrantLock lock;  // 空闲的内存块  private final Deque<ByteBuffer> free;  // 需要等待空闲内存块的事件  private final Deque<Condition> waiters;  /** Total available memory is the sum of nonPooledAvailableMemory and the number of byte buffers in free * poolableSize.  */  // 缓冲池还未分配的空闲内存,新申请的内存块就是从这里获取内存值  private long nonPooledAvailableMemory; // ...}

从 BufferPool 的成员可看出,缓冲池实际上由一个个 ByteBuffer 组成的,BufferPool 持有这些内存块,并保存在成员 free 中,free 的总大小由 totalMemory 作限制,而 nonPooledAvailableMemory 则表示还剩下缓冲池还剩下多少内存还未被分配。

当 Batch 的消息发送完毕后,就会将它持有的内存块归还到 free 中,以便后面的 Batch 申请内存块时不再创建新的 ByteBuffer,从 free 中取就可以了,从而避免了内存块被 JVM 回收的问题。

怎样深度剖析Kafka Producer的缓冲池机制

接下来跟大家一起分析申请内存和归还内存是如何实现的。

1、申请内存

申请内存的入口:

org.apache.kafka.clients.producer.internals.BufferPool#allocate

1)内存足够的情况

当用户请求申请内存时,如果发现 free 中有空闲的内存,则直接从中取:

if (size == poolableSize && !this.free.isEmpty()){
  return this.free.pollFirst(); 
}

这里的 size 即申请的内存大小,它等于 Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));

即如果你的消息大小小于 batchSize,则申请的内存大小为 batchSize,那么上面的逻辑就是如果申请的内存大小等于 batchSize 并且 free 不空闲,则直接从 free 中获取。

我们不妨想一下,为什么 Kafka 一定要申请内存大小等于 batchSize,才能从 free 获取空闲的内存块呢?

前面也说过,缓冲池的内存块大小是固定的,它等于 batchSize,如果申请的内存比 batchSize 还大,说明一条消息所需要存放的内存空间比内存块的内存空间还要大,因此不满足需求,不满组需求怎么办呢?我们接着往下分析:

// now check if the request is immediately satisfiable with the// memory on hand or if we need to blockint freeListSize = freeSize() * this.poolableSize;if (this.nonPooledAvailableMemory + freeListSize >= size) {  // we have enough unallocated or pooled memory to immediately  // satisfy the request, but need to allocate the buffer  freeUp(size);  this.nonPooledAvailableMemory -= size;}

freeListSize:指的是 free 中已经分配好并且已经回收的空闲内存块总大小;

nonPooledAvailableMemory:缓冲池还未分配的空闲内存,新申请的内存块就是从这里获取内存值;

this.nonPooledAvailableMemory + freeListSize:即缓冲池中总的空闲内存空间。

如果缓冲池的内存空间比申请内存大小要大,则调用 freeUp(size); 方法,接着将空闲的内存大小减去申请的内存大小。

private void freeUp(int size) {
  while (!this.free.isEmpty() && this.nonPooledAvailableMemory < size)
    this.nonPooledAvailableMemory += this.free.pollLast().capacity();
}

freeUp 这个方法很有趣,它的思想是这样的:

如果未分配的内存大小比申请的内存还要小,那只能从已分配的内存列表 free 中将内存空间要回来,直到 nonPooledAvailableMemory 比申请内存大为止。

2)内存不足的情况

在我的「Kafka Producer 异步发送消息居然也会阻塞?」这篇文章当中也提到了,当缓冲池的内存块用完后,消息追加调用将会被阻塞,直到有空闲的内存块。

阻塞等待的逻辑是怎么实现的呢?

// we are out of memory and will have to block
int accumulated = 0;
Condition moreMemory = this.lock.newCondition();try {
  long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);  this.waiters.addLast(moreMemory);
  // loop over and over until we have a buffer or have reserved
  // enough memory to allocate one
  while (accumulated < size) {
    long startWaitNs = time.nanoseconds();    long timeNs;    boolean waitingTimeElapsed;    try {
      waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
    } finally {
      long endWaitNs = time.nanoseconds();      timeNs = Math.max(0L, endWaitNs - startWaitNs);
      recordWaitTime(timeNs);    }    if (waitingTimeElapsed) {
      throw new TimeoutException("Failed to allocate memory within the configured max blocking time " + maxTimeToBlockMs + " ms.");
    }    remainingTimeToBlockNs -= timeNs;    // check if we can satisfy this request from the free list,
    // otherwise allocate memory
    if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {
      // just grab a buffer from the free list
      buffer = this.free.pollFirst();
      accumulated = size;    } else {
      // we'll need to allocate memory, but we may only get      // part of what we need on this iteration      freeUp(size - accumulated);      int got = (int) Math.min(size - accumulated, this.nonPooledAvailableMemory);      this.nonPooledAvailableMemory -= got;      accumulated += got;    }  }

以上源码的大致逻辑:

首先创建一个本次等待 Condition,并且把它添加到类型为 Deque 的 waiters 中(后面在归还内存中会唤醒),while 循环不断收集空闲的内存,直到内存比申请内存大时退出,在 while 循环过程中,调用 Condition#await 方法进行阻塞等待,归还内存时会被唤醒,唤醒后会判断当前申请内存是否大于 batchSize,如果等与 batchSize 则直接将归还的内存返回即可,如果当前申请的内存大于 大于 batchSize,则需要调用 freeUp 方法从 free 中释放空闲的内存出来,然后进行累加,直到大于申请的内存为止。

2、归还内存

申请内存的入口:

org.apache.kafka.clients.producer.internals.BufferPool#deallocate(java.nio.ByteBuffer, int)

public void deallocate(ByteBuffer buffer, int size) {
  lock.lock();
  try {
    if (size == this.poolableSize && size == buffer.capacity()) {
      buffer.clear();      this.free.add(buffer);
    } else {
      this.nonPooledAvailableMemory += size;
    }    Condition moreMem = this.waiters.peekFirst();
    if (moreMem != null)
      moreMem.signal();  } finally {
    lock.unlock();
  }}

归还内存块的逻辑比较简单:

如果归还的内存块大小等于 batchSize,则将其清空后添加到缓冲池的 free 中,即将其归还给缓冲池,避免了 JVM GC 回收该内存块。如果不等于呢?直接将内存大小累加到未分配并且空闲的内存大小值中即可,内存就无需归还了,等待 JVM GC 回收掉,最后唤醒正在等待空闲内存的线程。

怎样深度剖析Kafka Producer的缓冲池机制

经过以上的源码分析之后,给大家指出需要注意的一个问题,这会给 Producer 端带来严重的性能影响:

如果你的消息大小比 batchSize 还要大,则不会从 free 中循环获取已分配好的内存块,而是重新创建一个新的 ByteBuffer,并且该 ByteBuffer 不会被归还到缓冲池中(JVM GC 回收),如果此时 nonPooledAvailableMemory 比消息体还要小,还会将 free 中空闲的内存块销毁(JVM GC 回收),以便缓冲池中有足够的内存空间提供给用户申请,这些动作都会导致频繁 GC 的问题出现。

因此,需要根据业务消息的大小,适当调整 batch.size 的大小,避免频繁 GC。

上述就是小编为大家分享的怎样深度剖析Kafka Producer的缓冲池机制了,如果刚好有类似的疑惑,不妨参照上述分析进行理解。如果想知道更多相关知识,欢迎关注亿速云行业资讯频道。

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI