Skip to content

thread local

wangkaiwu edited this page Feb 9, 2023 · 2 revisions

线程局部变量

  • jdk 版本
  • netty 版本

jdk 版本

初始化

public class Thread implements Runnable {
    ThreadLocal.ThreadLocalMap threadLocals = null;
    ThreadLocal.ThreadLocalMap inheritableThreadLocals = null;
}
public class ThreadLocal<T> {
    static class ThreadLocalMap {
        private static final int INITIAL_CAPACITY = 16;

        ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
            table = new Entry[INITIAL_CAPACITY];
            int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
            table[i] = new Entry(firstKey, firstValue);
            size = 1;
            setThreshold(INITIAL_CAPACITY);
        }
    }
}

调用初始化的路径

public class ThreadLocal<T> {
    void createMap(Thread t, T firstValue) {
        t.threadLocals = new ThreadLocalMap(this, firstValue);
    }
}
public class ThreadLocal<T> {
    private T setInitialValue() {
        T value = initialValue();
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
        } else {
            createMap(t, value);
        }
        return value;
    }
}
public class ThreadLocal<T> {
    public T get() {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            // ..........
        }
        return setInitialValue();
    }
}

初始化或者赋值

public class ThreadLocal<T> {
    private T setInitialValue() {
        // 获取初始化的值
        T value = initialValue();
        Thread t = Thread.currentThread();
        // 获取局部变量的 map
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            // 此处是常规赋值
            map.set(this, value);
        } else {
            createMap(t, value);
        }
        return value;
    }
}
public class ThreadLocal<T> {
    static class ThreadLocalMap {
        private Entry[] table;

        private void set(ThreadLocal<?> key, Object value) {
            // table 的赋值
            Entry[] tab = table;
            int len = tab.length;
            // 在下面查看 threadLocalHashCode 的赋值
            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();
            }
        }
    }
}
public class ThreadLocal<T> {
    static class ThreadLocalMap {
        private static int nextIndex(int i, int len) {
            // 找下一个位置
            return ((i + 1 < len) ? i + 1 : 0);
        }
    }
}

附录

threadLocalHashCode 赋值

public class ThreadLocal<T> {
    // 通过原子自增获取存储索引
    private final int threadLocalHashCode = nextHashCode();
    private static AtomicInteger nextHashCode = new AtomicInteger();
    private static final int HASH_INCREMENT = 0x61c88647;

    private static int nextHashCode() {
        return nextHashCode.getAndAdd(HASH_INCREMENT);
    }
}

Entry 结构

public class ThreadLocal<T> {
    static class ThreadLocalMap {
        static class Entry extends WeakReference<ThreadLocal<?>> {
            Object value;

            Entry(ThreadLocal<?> k, Object v) {
                super(k);
                value = v;
            }
        }
    }
}
public class WeakReference<T> extends Reference<T> {
    public WeakReference(T referent) {
        super(referent);
    }
}
public abstract class Reference<T> {
    private T referent;
    volatile ReferenceQueue<? super T> queue;

    Reference(T referent) {
        this(referent, null);
    }

    Reference(T referent, ReferenceQueue<? super T> queue) {
        this.referent = referent;
        this.queue = (queue == null) ? ReferenceQueue.NULL : queue;
    }

    public T get() {
        return this.referent;
    }
}

小结

  • 线程本地变量存储在了 Thread 对象内
  • 查找变量的方式是开发地址法
  • 存储到 entry 是弱引用,有清除的可能

netty 版本

构造器

public class FastThreadLocalThread extends Thread {
    private final boolean cleanupFastThreadLocals;
    private InternalThreadLocalMap threadLocalMap;

    public FastThreadLocalThread() {
        cleanupFastThreadLocals = false;
    }
}

InternalThreadLocalMap 构造

public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {
    private static final int INDEXED_VARIABLE_TABLE_INITIAL_SIZE = 32;
    public static final Object UNSET = new Object();

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

    private static Object[] newIndexedVariableTable() {
        // 默认 32 的数组
        Object[] array = new Object[INDEXED_VARIABLE_TABLE_INITIAL_SIZE];
        Arrays.fill(array, UNSET);
        return array;
    }
}
class UnpaddedInternalThreadLocalMap {
    Object[] indexedVariables;

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

FastThreadLocal 构造

public class FastThreadLocal<V> {
    // 自增长
    private final int index;

    public FastThreadLocal() {
        index = InternalThreadLocalMap.nextVariableIndex();
    }
}
public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {
    static final AtomicInteger nextIndex = new AtomicInteger();

    public static int nextVariableIndex() {
        int index = nextIndex.getAndIncrement();
        return index;
    }
}

threadLocalMap 初始化

public class FastThreadLocalThread extends Thread {
    private InternalThreadLocalMap threadLocalMap;

    public final void setThreadLocalMap(InternalThreadLocalMap threadLocalMap) {
        this.threadLocalMap = threadLocalMap;
    }
}

以下是调用链

public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {
    private static InternalThreadLocalMap fastGet(FastThreadLocalThread thread) {
        InternalThreadLocalMap threadLocalMap = thread.threadLocalMap();
        if (threadLocalMap == null) {
            // 调用了默认构造方法
            thread.setThreadLocalMap(threadLocalMap = new InternalThreadLocalMap());
        }
        return threadLocalMap;
    }
}
public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {
    public static InternalThreadLocalMap get() {
        Thread thread = Thread.currentThread();
        if (thread instanceof FastThreadLocalThread) {
            // 执行 netty 版本获取
            return fastGet((FastThreadLocalThread) thread);
        } else {
            // 传统方式获取,开放地址法
            return slowGet();
        }
    }
}

ThreadLocal 的 get 和 set 都会调用上面的 get()

set 赋值过程

public class FastThreadLocal<V> {
    private final int index;

    public final void set(V value) {
        if (value != InternalThreadLocalMap.UNSET) {
            // 获取map
            InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
            // 真正的赋值
            setKnownNotUnset(threadLocalMap, value);
        } else {
            // 删除
            remove();
        }
    }

    private void setKnownNotUnset(InternalThreadLocalMap threadLocalMap, V value) {
        if (threadLocalMap.setIndexedVariable(index, value)) {
            // 说明是插入成功,否则是替换成功
            addToVariablesToRemove(threadLocalMap, this);
        }
    }
}

setIndexedVariable

可以初步看到,由于 index 原子自增,所以 hash 发生碰撞的概率较小

public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {
    Object[] indexedVariables;

    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;
        }
    }

    private void expandIndexedVariableTableAndSet(int index, Object value) {
        Object[] oldArray = indexedVariables;
        final int oldCapacity = oldArray.length;
        // 直接将新索引作为计算容量的基础
        int newCapacity = index;
        newCapacity |= newCapacity >>> 1;
        newCapacity |= newCapacity >>> 2;
        newCapacity |= newCapacity >>> 4;
        newCapacity |= newCapacity >>> 8;
        newCapacity |= newCapacity >>> 16;
        newCapacity++;
        // 生成新的数组
        Object[] newArray = Arrays.copyOf(oldArray, newCapacity);
        // 将超出 oldCapacity 的部分置为 UNSET
        Arrays.fill(newArray, oldCapacity, newArray.length, UNSET);
        // 放到新位置
        newArray[index] = value;
        // 替换原数组
        indexedVariables = newArray;
    }
}

addToVariablesToRemove

将变量添加到一个集合中

public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {
    // 默认值是 0 
    private static final int variablesToRemoveIndex = InternalThreadLocalMap.nextVariableIndex();

    private static void addToVariablesToRemove(InternalThreadLocalMap threadLocalMap, FastThreadLocal<?> variable) {
        // 获取第0个值
        Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);
        Set<FastThreadLocal<?>> variablesToRemove;
        if (v == InternalThreadLocalMap.UNSET || v == null) {
            // 说明 v 不存在
            variablesToRemove = Collections.newSetFromMap(new IdentityHashMap<FastThreadLocal<?>, Boolean>());
            threadLocalMap.setIndexedVariable(variablesToRemoveIndex, variablesToRemove);
        } else {
            variablesToRemove = (Set<FastThreadLocal<?>>) v;
        }

        variablesToRemove.add(variable);
    }
}

结论

相较于jdk的 local,netty 版有以下有点

  • 变量是强引用,不会清除
  • 采用索引自增的方式,不会发生hash碰撞
  • 变量被清除的时候可以收到回调
  • thread 运行 runnable 结束的时候,可以清理局部变量,避免在不同的 runnable 之间共享变量

Clone this wiki locally