FastThreadLocal 原理解析

news2024/12/29 9:53:53

FastThreadLocal

fastthreadlocal

每个 FastThread 包含一个 FastThreadLocalMap，每个 FastThreadLocalThread 中的多个 FastThreadLocal 占用不同的索引。
每个 InternalThreadLocalMap 的第一个元素保存了所有的 ThreadLocal 对象。之后的元素保存了每个 ThreadLocal 对应的 value

基本操作

get()

当前线程中

找到 ThreadLocal 的 index，value 加入 InternalThreadLocalMap 对应索引的位置

set()

当前线程中

找到 ThreadLocal 的 index，value 加入 InternalThreadLocalMap 对应索引的位置
将 ThreadLocal 加入当前线程 InternalThreadLocalMap 的第一个元素对应的集合

remove()

当前线程中

找到 ThreadLocal 的 index，InternalThreadLocalMap 中 index 对应的 value 置为 UNSET
将 ThreadLocal 从当前线程 InternalThreadLocalMap 的第一个元素集合中删除

关键设计点

兼容 ThreadLocal

当线程没有使用 FastThreadLocal 的时候，默认走 ThreadLocal 的逻辑。

初始大小

初始大小为 32

hash 算法

直接使用全局的自增，不存在Hash 冲突，以空间换时间

扩容条件是什么？如何扩容？

扩容条件：当前线程元素超出容量
扩容：元素数量扩展为向上取最近 2 次幂整数，比如，5 取8，31 取 64。

    public boolean setIndexedVariable(int index, Object value) {
        Object[] lookup = indexedVariables;
        // index 大于容量
        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;
        // 当小于 2的30次方时，容量扩展为向上取最近 2 次幂整数，比如，5 取8，31 取 64。
        if (index < ARRAY_LIST_CAPACITY_EXPAND_THRESHOLD) {
            newCapacity = index;
            newCapacity |= newCapacity >>>  1;
            newCapacity |= newCapacity >>>  2;
            newCapacity |= newCapacity >>>  4;
            newCapacity |= newCapacity >>>  8;
            newCapacity |= newCapacity >>> 16;
            newCapacity ++;
        } else {
            newCapacity = ARRAY_LIST_CAPACITY_MAX_SIZE;
        }

        Object[] newArray = Arrays.copyOf(oldArray, newCapacity);
        Arrays.fill(newArray, oldCapacity, newArray.length, UNSET);
        newArray[index] = value;
        indexedVariables = newArray;
    }

如何防止内存泄漏

自动：使用ftlt执行一个被FastThreadLocalRunnable wrap的Runnable任务，在任务执行完毕后会自动进行ftl的清理。

final class FastThreadLocalRunnable implements Runnable {
    private final Runnable runnable;

    private FastThreadLocalRunnable(Runnable runnable) {
        this.runnable = ObjectUtil.checkNotNull(runnable, "runnable");
    }

    @Override
    public void run() {
        try {
            runnable.run();
        } finally {
            FastThreadLocal.removeAll();
        }
    }

    static Runnable wrap(Runnable runnable) {
        return runnable instanceof FastThreadLocalRunnable ? runnable : new FastThreadLocalRunnable(runnable);
    }
}

手动： ftl和InternalThreadLocalMap都提供了remove方法，在合适的时候用户可以（有的时候也是必须，例如普通线程的线程池使用ftl）手动进行调用，进行显示删除。

使用

final class FastThreadLocalRunnable implements Runnable {
    private final Runnable runnable;

    private FastThreadLocalRunnable(Runnable runnable) {
        this.runnable = (Runnable)ObjectUtil.checkNotNull(runnable, "runnable");
    }

    public void run() {
        try {
            this.runnable.run();
        } finally {
            // 如果用的是 FastThreadLocalRunnable ，默认会做清理
            FastThreadLocal.removeAll();
        }

    }

    static Runnable wrap(Runnable runnable) {
        return (Runnable)(runnable instanceof FastThreadLocalRunnable ? runnable : new FastThreadLocalRunnable(runnable));
    }
}

存在什么问题

1、空间浪费，所有线程的 ThreadLocalMap 数组大小是一样的
比如，线程1 创建 100 个 ThreadLocal 对象。线程 1 里面有一个长度为 100 的数组。
此时，第二个线程需要调用 ThreadLocal 100 的 get 方法，第二个线程需要分配 100 个 Object 对象。

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

import java.util.ArrayList;
import java.util.List;

public class FastThreadLocalTest {

    private List<FastThreadLocal<String>> fastThreadLocals = new ArrayList<>();

    private List<ThreadLocal<String>> threadLocals = new ArrayList<>();

    void thread1Init() {
        new Thread (() -> {
            for (int i = 0; i < 31; i++) {
                ThreadLocal<String> threadLocal = new ThreadLocal<>();
                threadLocal.get();
                threadLocals.add(threadLocal);
            }
        }).start();

    }

    void thread2Init() {
        new Thread(() -> {
            threadLocals.get(threadLocals.size() - 1).get();
        });
    }

    void fastThread1Init() {
        new FastThreadLocalThread (() -> {
            for (int i = 0; i < 33; i++) {
                FastThreadLocal<String> fastThreadLocal = new FastThreadLocal<>();
                fastThreadLocal.get();
                fastThreadLocals.add(fastThreadLocal);
            }
        }).start();

    }

    void fastThread2Init() {
        new FastThreadLocalThread(() -> {
            fastThreadLocals.get(fastThreadLocals.size() - 1).get();
        });
    }

    public static void main(String[] args) {
        FastThreadLocalTest test = new FastThreadLocalTest();
        test.fastThread1Init();
        test.fastThread2Init();

        test.thread1Init();
        test.thread2Init();
    }
}