Netty源码分析第8章(高性能工具类FastThreadLocal和Recycler)---->第1节: FastThreadLocal的使用和建立

 

Netty源码分析第八章: 高性能工具类FastThreadLocal和Recycler

 

概述:

 

        FastThreadLocal咱们在剖析堆外内存分配的时候简单介绍过, 它相似于JDK的ThreadLocal, 也是用于在多线程条件下, 保证统一线程的对象共享, 只是netty中定义的FastThreadLocal, 性能要高于jdk的ThreadLocal, 具体缘由会在以后的小节进行剖析

 

        Recyler咱们应该也不会太陌生, 由于在以前章节中, 有好多地方使用了Recyler

 

        Recyler是netty实现的一个轻量级对象回收站, 不少对象在使用完毕以后, 并无直接交给gc去处理, 而是经过对象回收站将对象回收, 目的是为了对象重用和减小gc压力

 

        好比ByteBuf对象的回收, 由于ByteBuf对象在netty中会频繁建立, 而且会占用比较大的内存空间, 因此使用完毕后会经过对象回收站的方式进行回收, 已达到资源重用的目的

 

        这一章就对FastThreadLocal和Recyler两个并发工具类进行分析

 

第一节:FastThreadLocal的使用和建立

 

 

首先咱们看一个最简单的demo:

public class FastThreadLocalDemo { final class FastThreadLocalTest extends FastThreadLocal<Object>{ @Override protected Object initialValue() throws Exception { return new Object(); } } private final FastThreadLocalTest fastThreadLocalTest; public FastThreadLocalDemo(){ fastThreadLocalTest = new FastThreadLocalTest(); } public static void main(String[] args){ FastThreadLocalDemo fastThreadLocalDemo = new FastThreadLocalDemo(); new Thread(new Runnable() { @Override public void run() { Object obj = fastThreadLocalDemo.fastThreadLocalTest.get(); try { for (int i=0;i<10;i++){ fastThreadLocalDemo.fastThreadLocalTest.set(new Object()); Thread.sleep(1000); } }catch (Exception e){ e.printStackTrace(); } } }).start(); new Thread(new Runnable() { @Override public void run() { try { Object obj = fastThreadLocalDemo.fastThreadLocalTest.get(); for (int i=0;i<10;i++){ System.out.println(obj == fastThreadLocalDemo.fastThreadLocalTest.get()); Thread.sleep(1000); } }catch (Exception e){ } } }).start(); } }

这里首先声明一个内部类FastThreadLocalTest继承FastThreadLocal, 并重写initialValue方法, initialValue方法就是用来初始化线程共享对象的

而后声明一个成员变量fastThreadLocalTest, 类型就是内部类FastThreadLocalTest

在构造方法中初始化fastThreadLocalTest

 

main方法中建立当前类FastThreadLocalDemo的对象fastThreadLocalDemo

而后启动两个线程, 每一个线程经过fastThreadLocalDemo.fastThreadLocalTest.get()的方式拿到线程共享对象, 由于fastThreadLocalDemo是相同的, 因此fastThreadLocalTest对象也是同一个, 同一个对象在不一样线程中进行get()

第一个线程循环经过set方法修改共享对象的值

第二个线程则循环判断并输出fastThreadLocalTest.get()出来的对象和第一次get出来的对象是否相等

这里输出结果都true, 说明其余线程虽然不断修改共享对象的值, 但都不影响当前线程共享对象的值

这样就实现了线程共享的对象的功能

 

根据上述示例, 咱们剖析FastThreadLocal的建立

首先跟到FastThreadLocal的构造方法中:

public FastThreadLocal() { index = InternalThreadLocalMap.nextVariableIndex(); }

这里的index, 表明FastThreadLocal对象的一个下标, 每建立一个FastThreadLocal, 都会有一个惟一的自增的下标

跟到nextVariableIndex方法中:

public static int nextVariableIndex() { int index = nextIndex.getAndIncrement(); if (index < 0) { nextIndex.decrementAndGet(); throw new IllegalStateException("too many thread-local indexed variables"); } return index; }

这里只是获取nextIndex经过getAndIncrement()进行原子自增, 建立第一个FastThreadLocal对象时, nextIndex为0, 建立第二个FastThreadLocal对象时nextIndex为1, 以此类推, 第n次nextIndex为n-1, 如图所示

8-1-1

咱们回到demo中, 咱们看线程中的这一句:

Object obj = fastThreadLocalDemo.fastThreadLocalTest.get();

这里调用了FastThreadLocal对象的get方法, 做用是建立一个线程共享对象

咱们跟到get方法中:

public final V get() { return get(InternalThreadLocalMap.get()); }

这里调用了一个重载的get方法, 参数中经过InternalThreadLocalMap的get方法获取了一个InternalThreadLocalMap对象

咱们跟到InternalThreadLocalMap的get方法中, 分析其实如何获取InternalThreadLocalMap对象的

public static InternalThreadLocalMap get() { Thread thread = Thread.currentThread(); if (thread instanceof FastThreadLocalThread) { return fastGet((FastThreadLocalThread) thread); } else { return slowGet(); } }

这里首先拿到当前线程, 而后判断当前线程是否为FastThreadLocalThread线程, 一般NioEventLoop线程都是FastThreadLocalThread, 用于线程则不是FastThreadLocalThread

在这里, 若是FastThreadLocalThread线程, 则调用fastGet方法获取InternalThreadLocalMap, 从名字上咱们能知道, 这是一种效率极高的获取方式

若是不是FastThreadLocalThread线程, 则调用slowGet方式获取InternalThreadLocalMap, 一样根据名字, 咱们知道这是一种效率不过高的获取方式

咱们的demo并非eventLoop线程, 因此这里会走到slowGet()方法中

咱们首先剖析slowGet()方法:

private static InternalThreadLocalMap slowGet() { ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = UnpaddedInternalThreadLocalMap.slowThreadLocalMap; InternalThreadLocalMap ret = slowThreadLocalMap.get(); if (ret == null) { ret = new InternalThreadLocalMap(); slowThreadLocalMap.set(ret); } return ret; }

首先经过UnpaddedInternalThreadLocalMap.slowThreadLocalMap拿到一个ThreadLocal对象slowThreadLocalMap, slowThreadLocalMap是UnpaddedInternalThreadLocalMap类的一个静态属性, 类型是ThreadLocal类型

这里的ThreadLocal是jdk的ThreadLocal

而后经过slowThreadLocalMap对象的get方法, 获取一个InternalThreadLocalMap

若是第一次获取, InternalThreadLocalMap有多是null, 因此在if块中, new了一个InternalThreadLocalMap对象, 并设置在ThreadLocal对象中

由于netty实现的FastThreadLocal要比jdk的ThreadLocal要快, 因此这里的方法叫slowGet

回到InternalThreadLocalMap的get方法:

public static InternalThreadLocalMap get() { Thread thread = Thread.currentThread(); if (thread instanceof FastThreadLocalThread) { return fastGet((FastThreadLocalThread) thread); } else { return slowGet(); } }

咱们继续剖析fastGet方法, 一般EventLoop线程FastThreadLocalThread线程, 因此EventLoop线程执行到这一步的时候会调用fastGet方法

咱们跟进fastGet:

private static InternalThreadLocalMap fastGet(FastThreadLocalThread thread) { InternalThreadLocalMap threadLocalMap = thread.threadLocalMap(); if (threadLocalMap == null) { thread.setThreadLocalMap(threadLocalMap = new InternalThreadLocalMap()); } return threadLocalMap; }

首先FastThreadLocalThread对象直接经过threadLocalMap拿到threadLocalMap对象

若是threadLocalMap为null, 则建立一个InternalThreadLocalMap对象设置到FastThreadLocalThread的成员变量中

这里咱们能够知道FastThreadLocalThread对象中维护了一个InternalThreadLocalMap类型的成员变量, 能够直接经过threadLocalMap()方法获取该变量的值, 也就是InternalThreadLocalMap

咱们跟到InternalThreadLocalMap的构造方法中:

private InternalThreadLocalMap() { super(newIndexedVariableTable()); }

这里调用了父类的构造方法, 传入一个newIndexedVariableTable()

咱们跟到newIndexedVariableTable()中:

private static Object[] newIndexedVariableTable() { Object[] array = new Object[32]; Arrays.fill(array, UNSET); return array; }

这里建立一个长度为32的数组, 并为数组中的每个对象设置为UNSET, UNSET是一个Object的对象, 表示该下标的值没有被设置

回到InternalThreadLocalMap的构造方法, 再看其父类的构造方法:

UnpaddedInternalThreadLocalMap(Object[] indexedVariables) { this.indexedVariables = indexedVariables; }

这里初始化了一个数组类型的成员变量indexedVariables, 就是newIndexedVariableTable返回object的数组

这里咱们能够知道, 每一个InternalThreadLocalMap对象中都维护了一个Object类型的数组, 那么这个数组有什么做用呢?咱们继续往下剖析

回到FastThreadLocal的get方法中:

public final V get() { return get(InternalThreadLocalMap.get()); }

咱们剖析完了InternalThreadLocalMap.get()的相关逻辑, 再继续看重载的get方法:

public final V get(InternalThreadLocalMap threadLocalMap) { Object v = threadLocalMap.indexedVariable(index); if (v != InternalThreadLocalMap.UNSET) { return (V) v; } return initialize(threadLocalMap); }

首先看这一步:

Object v = threadLocalMap.indexedVariable(index);

这一步是拿到当前index下标的object, 其实也就是拿到每一个FastThreadLocal对象的绑定的线程共享对象

index是咱们刚才分析过, 是每个FastThreadLocal的惟一下标

咱们跟到indexedVariable方法中:

public Object indexedVariable(int index) { Object[] lookup = indexedVariables; return index < lookup.length? lookup[index] : UNSET; }

这里首先拿到indexedVariables, 咱们刚才分析过, indexedVariables是InternalThreadLocalMap对象中维护的数组, 初始大小是32

而后再return中判断当前index是否是小于当前数组的长度, 若是小于则获取当前下标index的数组元素, 不然返回UNSET表明没有设置的对象

这里咱们能够分析到, 其实每个FastThreadLocal对象中所绑定的线程共享对象, 是存放在threadLocalMap对象中的一个对象数组的中的, 数组中的元素的下标其实就是对应着FastThreadLocal中的index属性, 对应关系如图所示

8-1-2

回到FastThreadLocal重载的get方法:

public final V get(InternalThreadLocalMap threadLocalMap) { Object v = threadLocalMap.indexedVariable(index); if (v != InternalThreadLocalMap.UNSET) { return (V) v; } return initialize(threadLocalMap); }

根据以上逻辑, 咱们知道, 第一次获取对象v是只能获取到UNSET对象, 由于该对象并无保存在threadLocalMap中的数组indexedVariables中, 因此第一次获取在if判断中为false, 会走到initialize方法中

跟到initialize方法中:

private V initialize(InternalThreadLocalMap threadLocalMap) { V v = null; try { v = initialValue(); } catch (Exception e) { PlatformDependent.throwException(e); } threadLocalMap.setIndexedVariable(index, v); addToVariablesToRemove(threadLocalMap, this); return v; }

这里首先调用的initialValue方法, 这里的initialValue实际上走的是FastThreadLocal子类的重写initialValue方法

在咱们的demo中对应这个方法:

@Override protected Object initialValue() throws Exception { return new Object(); }

经过这个方法会建立一个线程共享对象

而后经过threadLocalMap对象的setIndexedVariable方法将建立的线程共享对象设置到threadLocalMap中维护的数组中, 参数为FastThreadLocal和建立的对象自己

跟到setIndexedVariable方法中:

public boolean setIndexedVariable(int index, Object value) { Object[] lookup = indexedVariables; if (index < lookup.length) { Object oldValue = lookup[index]; lookup[index] = value; return oldValue == UNSET; } else { expandIndexedVariableTableAndSet(index, value); return true; } }

这里首先判断FastThreadLocal对象的index是否超过数组indexedVariables的长度, 若是没有超过, 则直接经过下标设置新建立的线程共享对象, 经过这个操做, 下次获取该对象的时候就能够直接经过数组下标进行取出

若是index超过了数组indexedVariables的长度, 则经过expandIndexedVariableTableAndSet方法将数组扩容, 而且根据index的经过数组下标的方式将线程共享对象设置到数组indexedVariables中

以上就是线程共享对象的建立和获取的过程

 

上一节: Future和Promies

下一节: FastThreadLocal的set方法