ThreadLocal最常用的两个接口是set和get 最常见的应用场景为在线程上下文之间传递信息,使得用户不受复杂代码逻辑的影响
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
t.threadLocals;
我们使用set的时候实际上是获取Thread对象的threadLocals属性,把当前ThreadLocal当做参数然后调用其set(ThreadLocal,Object)方法来设值 threadLocals是ThreadLocal.ThreadLocalMap类型的
每个线程对象关联着一个ThreadLocalMap实例,主要是维护着一个Entry数组 Entry是扩展了WeakReference,提供了一个存储value的地方 一个线程对象可以对应多个ThreadLocal实例,一个ThreadLocal也可以对应多个Thread对象,当一个Thread对象和每一个ThreadLocal发生关系的时候会生成一个Entry,并将需要存储的值存储在Entry的value内
我们在看看ThreadLocalMap#set:
Entry[] tab = table;
int len = tab.length;
int i = key.threadLocalHashCode & (len-1);
for (Entry e = tab[i];
e != null;
e = tab[i = nextIndex(i, len)]) {
ThreadLocal k = e.get();
if (k == key) {
e.value = value;
return;
}
if (k == null) {
replaceStaleEntry(key, value, i);
return;
}
}
tab[i] = new Entry(key, value);
int sz = ++size;
if (!cleanSomeSlots(i, sz) && sz >= threshold)
rehash();
每个ThreadLocal实例都有一个唯一的threadLocalHashCode
初始值
上面首先根据threadLocalHashCode值计算出i,有下面两种情况会进入for循环:
threadLocalHashCode &(len-1)
对应的槽有内容,因此满足tab[i]!=null条件,进入for循环,如果满足条件且当前key不是当前threadlocal只能说明hash冲突了进入for循环会遍历tab数组,如果遇到以当前threadLocal为key的槽,即上面第(2)种情况,有则直接将值替换 如果找到了一个已经被回收的ThreadLocal对应的槽,也就是当key==null的时候表示之前的threadlocal已经被回收了,但是value值还存在,这也是ThreadLocal内存泄露的地方。碰到这种情况,则会引发替换这个位置的动作 如果上面两种情况都没发生,即上面的第(1)种情况,则新创建一个Entry对象放入槽中
private Entry getEntry(ThreadLocal key) {
int i = key.threadLocalHashCode & (table.length - 1);
Entry e = table[i];
if (e != null && e.get() == key)
return e;
else
return getEntryAfterMiss(key, i, e);
}
当命中的时候,也就是根据当前ThreadLocal计算出来的i恰好是当前ThreadLocal设置的值的时候,可以直接根据hashcode来计算出位置,当没有命中的时候,这里没有命中分为三种情况:
上面三种情况都会调用getEntryAfterMiss方法。调用getEntryAfterMiss方法会引发数组的遍历。
总结一下ThreadLocal的性能,一个线程对应多个ThreadLocal实例的场景中 在没有命中的情况下基本上一次hash就可以找到位置 如果发生没有命中的情况,则会引发性能会急剧下降,当在读写操作频繁的场景,这点将成为性能诟病。
两个线程操作同一object 对象,显然非线程安全,但是由于使用了 FTL, 线程安全!
结果表明内存地址不同,并非操作同一个 object!
让T1每1s 中新生成一个 object 对象 T2验证当前 object 是否与之前状态相同
显然,每个线程拿到的对象都是线程独享的! 某线程对变量的修改不影响其他线程! 通过对象隔离优化了程序性能!
创建时重写一下初始值方法
实际上在构造FastThreadLocal实例的时候就决定了这个实例的索引
index 为 private 且非 static, 说明每个实例都有该值
再看看索引的生成相关代码
index 从0开始计数
nextIndex是InternalThreadLocalMap父类的一个全局静态的AtomicInteger类型的对象,这意味着所有的FastThreadLocal实例将共同依赖这个指针来生成唯一的索引,而且是线程安全的 Netty重新设计了更快的FastThreadLocal,主要实现涉及
FastThreadLocalThread是Thread类的简单扩展,主要是为了扩展threadLocalMap属性
FastThreadLocal提供的接口和传统的ThreadLocal一致,主要是set和get方法,用法也一致 不同地方在于FastThreadLocal的值是存储在InternalThreadLocalMap这个结构里面的,传统的ThreadLocal性能槽点主要是在读写的时候hash计算和当hash没有命中的时候发生的遍历,我们来看看FastThreadLocal的核心实现
InternalThreadLocalMap实例和Thread对象一一对应 UnpaddedInternalThreadLocalMap维护着一个数组:
这个数组用来存储跟同一个线程关联的多个FastThreadLocal的值,由于FastThreadLocal对应indexedVariables的索引是确定的,因此在读写的时候将会发生随机存取,非常快。
另外这里有一个问题,nextIndex是静态唯一的,而indexedVariables数组是实例对象的,因此我认为随着FastThreadLocal数量的递增,这会造成空间的浪费
首先拿到当前线程,再判断是否为 FTL 线程快速获取否则慢速获取
首先会获取一个 ThreadLocal 变量
拿到 JDK 的 ThreadLocal 变量,用于给每个线程拿到InternalThreadLocalMap
变量,所以过程较慢,该方法称为 slowGet 可想而知!
由于在创建 ThreadLocal 时,并没有重写 initValue 方法,所以可能为 null
通过每个线程独享的 ThreadLocalMap 对象借助在 JVM 中每个 FTL 的唯一索引
所以不使用 new 而是直接复用
Netty使用
newObject
方法
再看看每个Recycler 的结构是如何的
看下 Stack对象 的 element 参数,一些默认处理器的数组,该数组实际存放对象池,每个处理器都包装了一个对象,handler 可被外部对象引用.,从而回收该对象
以上即为 Stack 对象所有成员变量.
下面回到Recycler的构造方法,看其入参.
该数值即为Stack 数组元素能有多少个
紧接着判断当前 size是否已到 maxcapacity,若达到上限,直接丢弃该对象即直接 return;否则继续判断 drop 处理器
首先判断,若该对象之前未被回收过,继续判断; 至今已经回收了多少个对象,其中 rm 为7,即111(二进制表示),即每隔8个对象,就会来此判断一次,将其与7进行与运算后,若不为0,则返回 true,表示只回收八分之一的对象.
继续回到 pushnow 的流程,接下来判断 size 是否等于数组的容量. 因为 element 是一个数组,并不是一开始就创建maxcapacity 容量大小,若容量不够了,则进行两倍大小扩容,再将其加入数组.
由前面 pushnow 进入同线程回收, pushlater 即进入异线程回收过程.
其类型就很神奇了,首先是个FTL,即每个线程都有一个 map,map 的key=stack 表示对于不同线程,对不同 stack 来说对应于不同的WOQ. 那么它为何要定义成一个 map 结构呢,假设有3个线程T1/2/3; T1创建的对象肯可能跑到T3中回收,T2中创建的对象也可能到了T3回收. 那么元素就是T1以及T2的WOQ.
假设当前在T2中,接下呢就通过get(this)拿到T1的WOQ,其中的 this 指的就是T1的 Stack. 然后若 queue==null,即表示T2从未回收过T1的对象,接下来开始判断 当前的即T2已经回收过的线程数 size,若不小于 mDQ,说明已经不能再回收其他线程的对象了! 给WOQ设置 dummy 标志,即对应下面的若下次看到一个线程标志了 dummy 直接return;什么也不做.
以上即为第一个过程,从FTL中拿一个Stack 对应的WOQ.
若之前没拿到呢,那就直接创建一个WOQ吧!
其中的 this 即为 stack,是在T1中维护的,thread 即表示当前线程T2. allocat就是为了给当前线程T2分配一个在T1中的Stack 对应的一个WOQ.
首先判断,T1中的 Stack还能否再分配LINK_capacity 个内存,若不能直接返回 null; 若可以,就直接 new 一个WOQ. 让我们具体看看其实现.
此函数意义为:该 Stack 允许外部线程给它缓存多少个对象 经过CAS操作设置该值为Stack 可为其他线程共享的回收对象的个数.
容量足够,则直接创建一个WOQ,下面来看看其数据结构.
一个链表结构.将其 handle与Link 进行分离,极大地提升了性能, 因为不必判断当前T2能否回收T1的对象,而只需判断当前的L ink 中是否有空的,则可直接将 hande 塞进去.因为在前面一次性的判断过,从T1中是否能批量分配这么多对象(以减少很多操作的频率).
使用同步,将WOQ插到Stack 的头部.
首先设置 上次回收 id
.
然后拿到尾指针,获取Link 的长度,若已经等于 link_capacity,说明已经不可写了; 继续判断 想办法看看T1是否还能再分配一个Link来保存待回收的对象. 不允许,则直接丢弃; 允许,则直接创建Link并重新赋值 tail 节点.
创建完后,拿到其写指针,即 tail 的长度(0).所以 tail 节点也已经又有了足够的存储空间,将 handle 追加进去.再将该 handle 的 stack 指针重置为 null,因为已经不属于原来的 stack 了. 最后,写指针+1.
本节分析若当前 stack 为空
若当前线程T1去获取对象,若 stack 中有对象,则直接拿出.T1所拥有的对象即为T1拥有的 stack 中的对象,若发现其中为空,会尝试与 和T1的 stack 关联的WOQ中的 T1创建的,但是在其他线程中去回收的对象.那么,T1中对象不足,就需要在其他线程中去回收.
其中的 cusor 指针即当前所需要回收的对象
若已经回收到则直接 return true.没有则重置两个指针,将 cusor 指向头结点,意味着准备从头开始回收.
接下来具体分析这段长代码
boolean scavengeSome() {
WeakOrderQueue prev;
// 先拿到 cusor
WeakOrderQueue cursor = this.cursor;
// cusor 节点无对象
if (cursor == null) {
prev = null;
// 指向头结点
cursor = head;
// 头结点依旧为空,已经没有与之关联的WOQ,直接返回 false.
if (cursor == null) {
return false;
}
} else {
prev = this.prev;
}
boolean success = false;
// 此处 do/while 循环只为去寻找与 stack 关联的WOQ,看看到底能不能碰到一个对象.
do {
// transfer 即为了将WOQ中的对象传输到 stack 中.成功获取则结束循环!
if (cursor.transfer(this)) {
success = true;
break;
}
// 没有回收成功,则看往 cusor 的下一个节点
WeakOrderQueue next = cursor.next;
// owner 为与当前WOQ关联的一个线程(对应图中的T4)
// 为空,说明T4已经不存在!随后即,做一些善后清理工作
if (cursor.owner.get() == null) {
// If the thread associated with the queue is gone, unlink it, after
// performing a volatile read to confirm there is no data left to collect.
// We never unlink the first queue, as we don't want to synchronize on updating the head.
// 判断节点中是否还有数据
if (cursor.hasFinalData()) {
// 就需要想办法将数据传输到 stack 中
for (;;) {
if (cursor.transfer(this)) {
success = true;
} else {
break;
}
}
}
// 处理完该节点后,即将其删除,通过传统的指针的删除方法
if (prev != null) {
prev.setNext(next);
}
// T4还存活,继续看后继节点.
} else {
prev = cursor;
}
cursor = next;
// cusor 为空时,诶就结束循环啦!
} while (cursor != null && !success);
this.prev = prev;
this.cursor = cursor;
return success;
}
// transfer as many items as we can from this queue to the stack, returning true if any were transferred
@SuppressWarnings("rawtypes")
boolean transfer(Stack<?> dst) {
// 先找到头结点
Link head = this.head;
if (head == null) {
// WOQ无数据,直接返回 false
return false;
}
// 说明当前Link 的所有数据已被取走.
if (head.readIndex == LINK_CAPACITY) {
if (head.next == null) {
return false;
}
this.head = head = head.next;
}
// 从该索引开始取对象
final int srcStart = head.readIndex;
int srcEnd = head.get();
// 当前Link 要传输到 Stack 的对象个数.
final int srcSize = srcEnd - srcStart;
if (srcSize == 0) {
return false;
}
// dst 为当前Stack
final int dstSize = dst.size;
// 将对象都传输到Stack 所需容量.
final int expectedCapacity = dstSize + srcSize;
// 所需容量大于Stack 的大小,
if (expectedCapacity > dst.elements.length) {
// 因此进行扩容
final int actualCapacity = dst.increaseCapacity(expectedCapacity);
// 可传输的最后一个对象
srcEnd = min(srcStart + actualCapacity - dstSize, srcEnd);
}
if (srcStart != srcEnd) {
final DefaultHandle[] srcElems = head.elements;
final DefaultHandle[] dstElems = dst.elements;
int newDstSize = dstSize;
for (int i = srcStart; i < srcEnd; i++) {
DefaultHandle element = srcElems[i];
//为0说明未被回收过
if (element.recycleId == 0) {
element.recycleId = element.lastRecycledId;
} else if (element.recycleId != element.lastRecycledId) {
throw new IllegalStateException("recycled already");
}
srcElems[i] = null;
if (dst.dropHandle(element)) {
// Drop the object.
continue;
}
element.stack = dst;
dstElems[newDstSize ++] = element;
}
if (srcEnd == LINK_CAPACITY && head.next != null) {
// Add capacity back as the Link is GCed.
reclaimSpace(LINK_CAPACITY);
this.head = head.next;
}
head.readIndex = srcEnd;
if (dst.size == newDstSize) {
return false;
}
dst.size = newDstSize;
return true;
} else {
// The destination stack is full already.
return false;
}
}