Netty對(duì)象池

import io.netty.util.Recycler;
public class Book {
    private String name;    private final Recycler.Handle<Book> recyclerHandle;
    Book(Recycler.Handle<Book> recyclerHandle) {        this.recyclerHandle = recyclerHandle;    }
    public void setName(String name) {        this.name = name;    }
    void recycle() {        recyclerHandle.recycle(this);    }}

import io.netty.util.Recycler;import io.netty.util.concurrent.FastThreadLocalThread;
import java.util.concurrent.locks.LockSupport;
public class Main {

    private static final Recycler<Book> BOOK = new Recycler<Book>() {        @Override        protected Book newObject(Handle<Book> handle) {            return new Book(handle);        }    };

    private static Book book1;    private static Book book2;    private static Book book3;
    public static void main(String[] args) throws Exception {

        new FastThreadLocalThread(() -> {            // book1,book2,book3都是從線程Thread-1中創(chuàng)建產(chǎn)生            book1 = BOOK.get();            book1.setName("Java");
            book2 = BOOK.get();            book2.setName("C");
            book3 = BOOK.get();            book3.setName("C++");            // book1在線程Thread-1中回收            book1.recycle();
                        new FastThreadLocalThread(() -> {                // book2在線程Thread-2中回收                book2.recycle();                LockSupport.park();// 讓線程不退出,便于觀察            }, "Thread-2").start();

            new FastThreadLocalThread(() -> {                // book3在線程Thread-3中回收                book3.recycle();                LockSupport.park();// 讓線程不退出,便于觀察            }, "Thread-3").start();

            LockSupport.park();// 讓線程不退出,便于觀察        }, "Thread-1").start();

        // 讓線程不退出,便于觀察        LockSupport.park();
    }}

jmap -dump:format=b,file=2021-heap.hprof 10624

io.netty.util.Recycler

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,                            interval, maxDelayedQueuesPerThread, delayedQueueInterval);    }
    @Override    protected void onRemoval(Stack<T> value) {        if (value.threadRef.get() == Thread.currentThread()) {            if (DELAYED_RECYCLED.isSet()) {                DELAYED_RECYCLED.get().remove(value);            }        }    }};

import io.netty.util.Recycler;import io.netty.util.concurrent.FastThreadLocalThread;

public class Main {
    private static final Recycler<Book> BOOK_CN = new Recycler<Book>() {        @Override        protected Book newObject(Handle<Book> handle) {            return new Book(handle);        }    };
    private static final Recycler<Book> BOOK_EN = new Recycler<Book>() {        @Override        protected Book newObject(Handle<Book> handle) {            return new Book(handle);        }    };
    public static void main(String[] args) throws Exception {
        new FastThreadLocalThread(() -> {            // 在同一個(gè)線程中使用2個(gè)Recycler實(shí)例,則會(huì)創(chuàng)建2個(gè)Stack            Book book1 = BOOK_CN.get();            Book book2 = BOOK_EN.get();
        }, "Thread-1").start();
    }}

public final T get() {    // 如果沒(méi)有啟用線程池,則每次獲取對(duì)象都要新創(chuàng)建對(duì)象.    if (maxCapacityPerThread == 0) {        return newObject((Handle<T>) NOOP_HANDLE);    }    // 每個(gè)線程獲取它自己對(duì)應(yīng)的Stack對(duì)象.    Stack<T> stack = threadLocal.get();    // 從Stack中'彈出'一個(gè)對(duì)象.    DefaultHandle<T> handle = stack.pop();    if (handle == null) {        // 如果Stack中沒(méi)有DefaultHandle對(duì)象,則新創(chuàng)建一個(gè)DefaultHandle        handle = stack.newHandle();        // 創(chuàng)建線程需要的那個(gè)對(duì)象        handle.value = newObject(handle);    }    // 返回線程需要的那個(gè)對(duì)象    return (T) handle.value;}

DefaultHandle<T> pop() {    int size = this.size;    if (size == 0) {        // 這個(gè)地方后面會(huì)說(shuō)        if (!scavenge()) {            return null;        }        size = this.size;        if (size <= 0) {            return null;        }    }    size --;    // 從elements數(shù)組中拿出最后一個(gè)元素    DefaultHandle ret = elements[size];    elements[size] = null;
    this.size = size;
    if (ret.lastRecycledId != ret.recycleId) {        throw new IllegalStateException("recycled multiple times");    }    ret.recycleId = 0;    ret.lastRecycledId = 0;    // 將元素直接返回    return ret;}

public class Book {
    private String name;
    private final Recycler.Handle<Book> recyclerHandle;
    Book(Recycler.Handle<Book> recyclerHandle) {        this.recyclerHandle = recyclerHandle;    }
    public void setName(String name) {        this.name = name;    }
    // 回收方法    void recycle() {        // 實(shí)際調(diào)用io.netty.util.Recycler.DefaultHandle#recycle方法        recyclerHandle.recycle(this);    }
}

@Overridepublic void recycle(Object object) {    if (object != value) {        throw new IllegalArgumentException("object does not belong to handle");    }  // 得到相應(yīng)的Stack    // 每個(gè)DefaultHandle有且僅屬于一個(gè)Stack    Stack<?> stack = this.stack;    if (lastRecycledId != recycleId || stack == null) {        throw new IllegalStateException("recycled already");    }  // 將DefaultHandle對(duì)象放入Stack中    stack.push(this);}

void push(DefaultHandle<?> item) {    // 得到當(dāng)前線程    Thread currentThread = Thread.currentThread();        // 如果當(dāng)前線程和Stack對(duì)應(yīng)的線程是同一個(gè)線程    // 每個(gè)Stack有且只屬于一個(gè)線程.    if (threadRef.get() == currentThread) {        pushNow(item);    } else {        // 當(dāng)前線程和Stack對(duì)應(yīng)的線程不是同一個(gè)線程        pushLater(item, currentThread);    }}

private void pushNow(DefaultHandle<?> item) {    if (item.recycleId != 0 || !item.compareAndSetLastRecycledId(0, OWN_THREAD_ID)) {        throw new IllegalStateException("recycled already");    }    item.recycleId = OWN_THREAD_ID;
    int size = this.size;    // dropHandle方法用來(lái)控制放入Stack的速率,有點(diǎn)類(lèi)似流控,稍后再說(shuō)    if (size >= maxCapacity || dropHandle(item)) {        return;    }    if (size == elements.length) {        elements = Arrays.copyOf(elements, min(size << 1, maxCapacity));    }  // 直接將元素放入elements數(shù)組的最后    elements[size] = item;    this.size = size + 1;}

// 1.先將Queue鏈接到鏈表上// 2.將item添加到Queue#Linkprivate void pushLater(DefaultHandle<?> item, Thread thread) {
    // DELAYED_RECYCLED是FastThreadLocal類(lèi)型, 則每個(gè)線程都有一個(gè)DELAYED_RECYCLED.    // 每個(gè)線程會(huì)維護(hù) Stack -> WeakOrderQueue 映射關(guān)系, 即每個(gè)線程會(huì)'協(xié)助'其他線程存儲(chǔ)它的對(duì)象.    Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();    WeakOrderQueue queue = delayedRecycled.get(this);    if (queue == null) {        if (delayedRecycled.size() >= maxDelayedQueues) {            delayedRecycled.put(this, WeakOrderQueue.DUMMY);            return;        }        // 創(chuàng)建queue,同時(shí)鏈接到鏈表上        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);}

boolean dropHandle(DefaultHandle<?> handle) {    // 1.如果當(dāng)前handle之前被回收過(guò),那么此次也會(huì)被回收    // 2.如果當(dāng)前handle之前沒(méi)有回收過(guò),那么默認(rèn)每隔8個(gè)回收一個(gè),防止Stack的DefaultHandle[]數(shù)組發(fā)生爆炸性的增長(zhǎng).    if (!handle.hasBeenRecycled) {        if ((++handleRecycleCount & ratioMask) != 0) {            // Drop the object.            return true;        }        handle.hasBeenRecycled = true; // 標(biāo)記元素被回收    }    return false;}

DefaultHandle<T> pop() {    int size = this.size;    if (size == 0) {// 如果elements中沒(méi)有可用元素        if (!scavenge()) {// 從其他WeakOrderQueue中轉(zhuǎn)移數(shù)據(jù)到當(dāng)前stack#elements中            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;}