Netty 高性能之道 - Recycler 对象池的复用

作者: 莫那一鲁道 | 来源:发表于2018-03-22 01:33 被阅读69次

Recycler 设计图

前言

我们知道，Java 创建一个实例的消耗是不小的，如果没有使用栈上分配和 TLAB，那么就需要使用 CAS 在堆中创建对象。所以现在很多框架都使用对象池。Netty 也不例外，通过重用对象，能够避免频繁创建对象和销毁对象带来的损耗。

来看看具体实现。

1. Recycler 抽象类简介

该类 doc：

Light-weight object pool based on a thread-local stack.
基于线程局部堆栈的轻量级对象池。

该类是个容器，内部主要是一个 Stack 结构。当需要使用一个实例时，就弹出，当使用完毕时，就清空后入栈。

该类有 2 个主要方法：

1. public final T get() // 从 threadLocal 中取出 Stack 中首个 T 实例。
2. protected abstract T newObject(Handle<T> handle) // 当 Stack 中没有实例的时候，创建一个实例返回。

该类有 4 个内部接口 / 内部类：

// 定义 handler 回收实例
public interface Handle<T> {
    void recycle(T object);
}

// Handle 的默认实现，可以将实例回收，放入 stack。
static final class DefaultHandle<T> implements Handle<T>

// 存储对象的数据结构。对象池的真正的 “池”
static final class Stack<T>

// 多线程共享的队列
private static final class WeakOrderQueue

// 队列中的链表结构，用于存储多线程回收的实例
private static final class Link extends AtomicInteger

实现线程局部缓存的 FastThreadLocal：

private final FastThreadLocal<Stack<T>> threadLocal = new FastThreadLocal<Stack<T>>() {
    @Override
    protected Stack<T> initialValue() {
        return new Stack<T>(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor,
                ratioMask, maxDelayedQueuesPerThread);
    }

    @Override
    protected void onRemoval(Stack<T> value) {
        if (value.threadRef.get() == Thread.currentThread()) {
           if (DELAYED_RECYCLED.isSet()) {
               DELAYED_RECYCLED.get().remove(value);
           }
        }
    }
};

核心方法 get 操作

public final T get() {
    if (maxCapacityPerThread == 0) {
        return newObject((Handle<T>) NOOP_HANDLE);
    }
    Stack<T> stack = threadLocal.get();
    DefaultHandle<T> handle = stack.pop();
    if (handle == null) {
        handle = stack.newHandle();
        handle.value = newObject(handle);
    }
    return (T) handle.value;
}

核心方法 DefaultHandle 的 recycle 操作

public void recycle(Object object) {
    if (object != value) {
        throw new IllegalArgumentException("object does not belong to handle");
    }
    stack.push(this);
}

2. Netty 中的使用范例

io.netty.channel.ChannelOutboundBuffer.Entry 类

示例代码如下：

// 实现了 Recycler 抽象类
private static final Recycler<Entry> RECYCLER = new Recycler<Entry>() {
    protected Entry newObject(Handle<Entry> handle) {
        return new Entry(handle);
    }
};

// 创建实例
Entry entry = RECYCLER.get();
// doSomeing......
// 归还实例
entry.recycle();

从上面的 get 方法，我们知道，最终会从 threadLocal 取出 Stack，从 Stack 中弹出 DefaultHandle 对象（如果没有就创建一个），然后调用我们重写的 newObject 方法，将创建的对象和 handle 绑定。最后返回这个对象。

当调用 entry.recycle() 方法的时候，实际会调用 DefaultHandle 的 recycle 方法。我们看看该方法实现：

public void recycle(Object object) {
    if (object != value) {
        throw new IllegalArgumentException("object does not belong to handle");
    }
    stack.push(this);
}

这里的 value 就是 get 方法中赋值的。如果不相等，就抛出异常。反之，将 handle 入栈 stack。注意：这里并没有对 value 做任何处理，只是在 Entry 内部做了清空处理。所以，这个 handle 和 handle 绑定的对象就保存在了 stack 中。

下次再次调用 get 时，就可以直接从该 threadLocal 中取出 handle 和 handle 绑定的 value了。完成了一次完美的对象池的实践。也就是说，一个 handle 绑定一个实例。而这个 handle 还是比较轻量的。

从这里可以看出，Stack 就是真正的 “池子”。我们就看看这个池子的内部实现。

而这个 stack 对外常用的方法的 pop 和 push。我们就来看看这两个方法。

3. pop 方法

代码如下：

DefaultHandle<T> pop() {
    int size = this.size;
    if (size == 0) {
        if (!scavenge()) {
            return null;
        }
        size = this.size;
    }
    size --;
    DefaultHandle ret = elements[size];
    elements[size] = null;
    if (ret.lastRecycledId != ret.recycleId) {
        throw new IllegalStateException("recycled multiple times");
    }
    ret.recycleId = 0;
    ret.lastRecycledId = 0;
    this.size = size;
    return ret;
}

逻辑如下：

拿到这个 Stack 的长度，实际上，这个 Stack 就是一个 DefaultHandle 数组。
如果这个长度是 0，没有元素了，就调用 scavenge 方法尝试从 queue 中转移一些数据到 stack 中。scavenge 方法待会详细再讲。
重置 size 属性和其余两个属性。返回实例。

这个方法除了 scavenge 之外，还是比较简单的。

4. push 方法

代码如下：

 void push(DefaultHandle<?> item) {
    Thread currentThread = Thread.currentThread();
    if (threadRef.get() == currentThread) { 
        pushNow(item);
    } else { 
        pushLater(item, currentThread);
    }
}

当一个对象使用 pop 方法取出来之后，可能会被别的线程使用，这时候，如果是你，你怎么处理呢？

先看看当前线程的处理：

看看 pushNow 方法：

private void pushNow(DefaultHandle<?> item) {
    if ((item.recycleId | item.lastRecycledId) != 0) {
        throw new IllegalStateException("recycled already");
    }
    item.recycleId = item.lastRecycledId = OWN_THREAD_ID;

    int size = this.size;
    if (size >= maxCapacity || dropHandle(item)) {
        return;
    }
    if (size == elements.length) {
        elements = Arrays.copyOf(elements, min(size << 1, maxCapacity));
    }

    elements[size] = item;
    this.size = size + 1;
}

该方法主要逻辑如下：

如果 Stack 大小已经大于等于最大容量或者这个 handle 在容器里了，就不做回收了。
如果数组满了，扩容一倍，最大 4096（默认）。
size + 1。

看看 dropHandle 方法的实现：

boolean dropHandle(DefaultHandle<?> handle) {
    // 没有被回收过
    if (!handle.hasBeenRecycled) {
        // 第一次是 -1，++ 之后变为0，取余7。其实如果正常情况下，结果应该都是0。
        // 如果下面的判断不是0 的话，那么已经归还。这个对象就没有必要重复归还。
        // 直接丢弃。
        if ((++handleRecycleCount & ratioMask) != 0) {
            // Drop the object.
            return true;
        }
        // 改为被回收过，下次就不会进入了
        handle.hasBeenRecycled = true;
    }
    // 删除失败
    return false;
}

已经写了注释，就不再过多解释。

可以看到，pushNow 方法还是很简单的。由于在当前线程里，只需要还原到 Stack 的数组中就好了。

关键是：如果是其他的线程做回收操作，该怎么办？

5. pushLater 方法（多线程回收如何操作）

先说说 Netty 的解决办法和思路：

每个线程都有一个 Stack 对象，每个线程也都有一个软引用 Map，键为 Stack，值是 queue。
线程每次从 local 中获取 Stack 对象，再从 Stack 中取出实例。如果取不到，尝试从 queue 取，也就是从queue 中的 Link 中取出，并销毁 Link。
但回收的时候，可能就不是原来的那个线程了，由于回收时使用的还是原来的 Stack，所以，需要考虑这个实例如何才能正确的回收。
这个时候，就需要 Map 出场了。创建一个 queue 关联这个 Stack，将数据放到这个 queue 中。等到持有这个 Stack 的线程想拿数据了，就从 Stack 对应的 queue 中取出。
看出来了吗？只有一个线程持有唯一的 Stack，其余的线程只持有这个 Stack 关联的 queue。因此，可以说，这个 queue 是两个线程共享的。除了 Stack 自己的线程外，其余的线程的归还都是放到自己的queue 中。
这个 queue 是无界的。内部的 Link 是有界的。每个线程对应一个 queue。
这些线程的 queue 组成了链表。

具体如下图所示：

Recycler 设计图

看完了设计，再看看代码实现：

pushLater 方法

private void pushLater(DefaultHandle<?> item, Thread thread) {
    // 每个 Stack 对应一串 queue，找到当前线程的 map
    Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
    // 查看当前线程中是否含有这个 Stack 对应的队列
    WeakOrderQueue queue = delayedRecycled.get(this);
    if (queue == null) {// 如果没有
        // 如果 map 长度已经大于最大延迟数了，则向 map 中添加一个假的队列
        if (delayedRecycled.size() >= maxDelayedQueues) {// 8
            delayedRecycled.put(this, WeakOrderQueue.DUMMY);
            return;
        }
        // 如果长度不大于最大延迟数，则尝试创建一个queue，链接到这个 Stack 的 head 节点前（内部创建Link）
        if ((queue = WeakOrderQueue.allocate(this, thread)) == null) {
            // drop object
            return;
        }
        delayedRecycled.put(this, queue);
    } else if (queue == WeakOrderQueue.DUMMY) {
        // drop object
        return;
    }

    queue.add(item);
}

该方法步骤如下：

从 threadLcoal 中取出当前线程的 Map，尝试从 Map 中取出 Stack 映射的 queue。
如果没有，就调用 WeakOrderQueue.allocate(this, thread) 方法创建一个。然后将这个 Stack 和 queue 绑定。
将实例添加到这个 queue 中。

我们主要关注如何 allocate 方法，关键方法 newQueue：

@1 
static WeakOrderQueue newQueue(Stack<?> stack, Thread thread) {
    WeakOrderQueue queue = new WeakOrderQueue(stack, thread);
    stack.setHead(queue);
    return queue;
}

@2 
private WeakOrderQueue(Stack<?> stack, Thread thread) {
    head = tail = new Link();
    owner = new WeakReference<Thread>(thread);
    availableSharedCapacity = stack.availableSharedCapacity;
}

@3
private static final class Link extends AtomicInteger {
    private final DefaultHandle<?>[] elements = new DefaultHandle[LINK_CAPACITY];
    private int readIndex;
    private Link next;
}

@4
synchronized void setHead(WeakOrderQueue queue) {
    queue.setNext(head);
    head = queue;
}

代码1，2，3，4。

调用 WeakOrderQueue 构造方法，传入 stack 和 thread。
创建一个 Link 对象，赋值给链表中的 head 和 tail。
Lind 的构造函数，也是一个链表。其中包含了保存实例的 Handle 数组，默认 16.
将这个新的 queue 设置为该 stack 的 head 节点。

其中，有一个需要注意的地方就是 owner = new WeakReference<Thread>(thread)，使用了弱引用，当这个线程对象被 GC 后，这个 owner 也会变为 null，就可以像 threadLoca 一样对该引用进行判 null，来检查这个线程对象是否回收了。

再看看如何添加进 queue 中的：

void add(DefaultHandle<?> handle) {
    handle.lastRecycledId = id;
    Link tail = this.tail;
    int writeIndex;
    if ((writeIndex = tail.get()) == LINK_CAPACITY) {
        if (!reserveSpace(availableSharedCapacity, LINK_CAPACITY)) {
            // Drop it.
            return;
        }
        this.tail = tail = tail.next = new Link();
        writeIndex = tail.get();
    }
    tail.elements[writeIndex] = handle;
    handle.stack = null;
    tail.lazySet(writeIndex + 1);
}

首先，拿到这个 queue 的 tail 节点，如果这个 tiail 节点满了，查看是否还有共享空间，如果没了，就丢弃这个实例。
反之，则新建一个 Link，追加到 tail 节点的尾部。然后，将数据插入新 tail 的数组。然后，将这个 handle 的 stack 属性设置成 null，表示这个 handle 不属于任何 statck 了，其他 stack 都可以使用。

数据放进去了，怎么取出来呢？

6. scavenge 方法

我们刚刚留了这个方法，现在可以开始讲了。代码如下：

boolean scavenge() {
    // continue an existing scavenge, if any
    // 清理成功后，stack 的 size 会变化
    if (scavengeSome()) {
        return true;
    }

    // reset our scavenge cursor
    prev = null;
    cursor = head;
    return false;
}

主要调用的是 scavengeSome 方法，返回 true 表示将 queue 中的数据转移成功。看看该方法。

boolean scavengeSome() {
    WeakOrderQueue prev;
    WeakOrderQueue cursor = this.cursor;
    if (cursor == null) {
        prev = null;
        cursor = head;
        if (cursor == null) {
            return false;
        }
    } else {
        prev = this.prev;
    }
    boolean success = false;
    do {
        // 将 head queue 的实例转移到 this stack 中
        if (cursor.transfer(this)) {
            success = true;
            break;
        }
        // 如果上面失败，找下一个节点
        WeakOrderQueue next = cursor.next;
        // 如果当前线程被回收了，
        if (cursor.owner.get() == null) {
          // 只要最后一个节点还有数据，就一直转移
            if (cursor.hasFinalData()) {
                for (;;) {
                    if (cursor.transfer(this)) {
                        success = true;
                    } else {
                        break;
                    }
                }
            }
            if (prev != null) {
                prev.setNext(next);
            }
        } else {
            prev = cursor;
        }
        cursor = next;
    } while (cursor != null && !success);
    // 转移成功之后，将 cursor 重置
    this.prev = prev;
    this.cursor = cursor;
    return success;
}

方法还是挺长的。我们拆解一下：

拿到这个 stack 的 queue，调用这个 queue 的 transfer 方法，如果成功，结束循环。
如果 queue 所在的线程被回收了，就将这个线程对应的 queue 中的所有数据全部转移到 stack 中。

可以看到，最重要的还是 transfer 方法。然而该方法更长，就不贴代码了，说说主要逻辑，有兴趣可以自己看看，逻辑如下：

拿到这个 queue 的 head 节点，也就是 Link。如果 head 是 null，取出 next。
循环 Link 中的实例，将其赋值到 stack 数组中。并将刚刚 handle 置为 null 的 stack 属性赋值。
最后，将 statck 的 size 属性更新。

其中有一个疑问：为什么在其他线程插入 Link 时将 handle 的 stack 的属性置为 null？在取出时，又将 handle 的 stack 属性恢复。

答：因为如果 stack 被用户手动置为 null，而容器中的 handle 还持有他的引用的话，就无法回收了。同时 Map 也使用了软引用map，当 stack 没有了引用被 GC 回收时，对应的 queue 也就被回收了。避免了内存泄漏。实际上，在之前的 Recycler 版本中，确实存在内存泄漏的情况。

该方法的主要目的就是将 queue 所属的 Link 中的数据转移到 stack 中。从而完成多线程的最终回收。

总结

Netty 并没有使用第三方库实现对象池，而是自己实现了一个相对轻量的对象池。通过使用 threadLocal，避免了多线程下取数据时可能出现的线程安全问题，同时，为了实现多线程回收同一个实例，让每个线程对应一个队列，队列链接在 Stack 对象上形成链表，这样，就解决了多线程回收时的安全问题。同时，使用了软引用的map 和软引用的 thradl 也避免了内存泄漏。

在本次的源码阅读中，确实收获很大。再回顾以下 Recycler 的设计图吧。设计的真的非常好。

Recycler 设计图

Netty 高性能之道 - Recycler 对象池的复用

前言

1. Recycler 抽象类简介

2. Netty 中的使用范例

3. pop 方法

4. push 方法

5. pushLater 方法（多线程回收如何操作）

6. scavenge 方法

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Netty

netty