当前位置:   article > 正文

kafka源码解析(4)生产者解析之内存池_kafka we are out of memory and will have to block

kafka we are out of memory and will have to block

使用内存池的好处

众所周知,jvm的GC会有不小的时间损耗,stop the world会严重影响kafka 生产者的消息发送。如果我们每次消息发送后,对中间生成的实例不做任何处理留给JVM,可能会造成严重的后果。因此Kafka生产者用内存池这种东西来循环利用中间生成的消息缓存bytebuffer。

依据我的理解,就是:标准大小消息的bytebuffer,用过一次后接着用装下一条消息。对于非标准大小的消息,用另外手段分配。

对于大消息手动使用赋值null对于GC有无帮助?看到下面的说明
java对象不再使用时,为什么要赋值为null

直接覆写bytebuffer确实快。但是看完内存池的代码,在这里BB两句:对于大消息,kafka还是要依靠JVM GC。那为什么不在缓存池外直接new bytebuffer,使用完后自己赋值null释放呢,非要来内存池抢东西。

BufferPool

接着看kafka的内存池吧,RecordAccumulator中的append方法是把消息添加进batches里面的相应dq,在添加前需要把消息放进bytebuffer里面,bytebuffer从free里面allocate方法要来的。多个生产者会有线程安全问题,比如这边计算可用内存后别人给用掉了,所以allocate时候要锁住Bufferpool

buffer = free.allocate(size, maxTimeToBlock);
  • 1

free就是内存池,几个比较重要的属性和解释如下

	//总大小 默认32M
	private final long totalMemory;
    
    //标准批次大小
    private final int poolableSize;
    
    //可用bytebuffer(回收的,一开始没有)
    private final Deque<ByteBuffer> free;
    
    //因为内存不足等待的线程队列
    private final Deque<Condition> waiters;
	 
	//totalMemory - poolableSize*free.size() - 使用中的butebuffer总大小
    private long nonPooledAvailableMemory;
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

情景1 : free队列有bytebuffer且size是标准大小

allocate方法中,首先会对批次的大小做一个判断,如果大于totalMemory直接报错。

free队列有bytebuffer且size是标准大小,直接安排覆写bytebuffer

if (size == poolableSize && !this.free.isEmpty())
                return this.free.pollFirst();
  • 1
  • 2

情景2 :size大于标准且可用内存能满足size

然后 this.nonPooledAvailableMemory + freeListSize 得出可用内存,在能满足size的条件下,freeUp(size)后,跳出try,走 safeAllocateByteBuffer(size),随便找一块内存安排上(这块内存就无法回收到free队列了,只能靠JVM回收),同时nonPooledAvailableMemory扣减size大小

if (this.nonPooledAvailableMemory + freeListSize >= size) {
                freeUp(size);
                this.nonPooledAvailableMemory -= size;
            } 
  • 1
  • 2
  • 3
  • 4

freeUp看下,nonPooledAvailableMemory 顶不住的话,不断释放free来补nonPooledAvailableMemory

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

情景3:(size大于标准但可用内存不能满足size)或者(size大于标准但是free空了)

这就麻烦了,得在线等在使用的bytebuffer释放

用accumulated记录搞来的内存

先把当前线程制作成Condition放入wait队列
关于Condition

// 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;
                        }
                    }
                    // Don't reclaim memory on throwable since nothing was thrown
                    accumulated = 0;
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43

情景3-1 size大于标准但是free空了

看上面的代码,发现free不空直接安排了

情景3-2 size大于标准但可用内存不能满足size

看上面的代码,freeUp(size - accumulated),

有可能这还不行,那就算下 Math.min(size - accumulated, this.nonPooledAvailableMemory),看看又榨取了多少内存
accumulated += got, 把榨取来的加到accumulated

跳出后waiters.remove(moreMemory),把刚满足的这个Condition去掉

装备回收

每次allocate成功后,顺手叫醒等待在waiter中的最早的线程,并且解锁内存池

!(this.nonPooledAvailableMemory == 0 && this.free.isEmpty())判断还有能否榨取的内存,没有就没必要唤醒等待者了。

try {
                if (!(this.nonPooledAvailableMemory == 0 && this.free.isEmpty()) && !this.waiters.isEmpty())
                    this.waiters.peekFirst().signal();
            } finally {
                // Another finally... otherwise find bugs complains
                lock.unlock();
            }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

如果buffer使用过了,要从内存池里面deallocate这个buffer,注意这个append的finally的dellocate针对的不是放入dq这个情况哦,成功放进dq后,buffer指向null

finally里面的deallocate针对是在自建batch前成功塞入dq的情况

if (buffer != null)
                free.deallocate(buffer);
            	appendsInProgress.decrementAndGet();
  • 1
  • 2
  • 3

deallocate 方法也看下,size = 标准就安排进内存池

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();
        }
    }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

看完有几点需要记录下:

nonPooledAvailableMemory 是个long,代表还能从堆空间拿多少内存,其他文章总说未申请空间感觉怪怪的,bufferpool实例化时并不是占用上32MB空间,上限是侵占堆空间32MB

ReentrantLock锁后面需要了解下

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/我家自动化/article/detail/626074
推荐阅读
相关标签
  

闽ICP备14008679号