首页 > 面试高频ThreadLocal
头像
mlpan
编辑于 2021-05-07 10:30
+ 关注

面试高频ThreadLocal

ThreadLocal底层数据结构

相关的CSDN博客:

ThreadLocal底层

ThreadLocal ,也叫线程本地变量,ThreadLocal为变量在每个线程中都创建了所使用的的变量副本。使用起来都是在线程的本地工作内存中操作,并且提供了set和get方法来访问拷贝过来的变量副本。底层也是封装了ThreadLocalMap集合类来绑定当前线程和变量副本的关系,各个线程独立并且访问安全!

底层:

其实ThreadLocal里面封装了ThreadLocalMap集合类来绑定当前线程和变量副本的关系。

ThreadLocalMap其实就是利用数组进行实现的。跟HashMap相似

根据key.threadLocalHashCode & (table.length - 1);获取下标值,然后获取到数组的值

static class ThreadLocalMap {

    static class Entry extends WeakReference<ThreadLocal> {
        Object value;
        Entry(ThreadLocal k, Object v) {
            super(k);
            value = v;
        }
    }
    // 初始容量
    private static final int INITIAL_CAPACITY = 16;

    // 核心数组
    private Entry[] table;

    private int size = 0;

    private int threshold; // Default to 0

    private void setThreshold(int len) {
        threshold = len * 2 / 3;
    }
    // i的下一个下标,其实就是保证循环
    private static int nextIndex(int i, int len) {
        return ((i + 1 < len) ? i + 1 : 0);
    }
    // i的上一个下标,其实就是保证循环
    private static int prevIndex(int i, int len) {
        return ((i - 1 >= 0) ? i - 1 : len - 1);
    }


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


    private ThreadLocalMap(ThreadLocalMap parentMap) {
        Entry[] parentTable = parentMap.table;
        int len = parentTable.length;
        setThreshold(len);
        table = new Entry[len];

        for (int j = 0; j < len; j++) {
            Entry e = parentTable[j];
            if (e != null) {
                ThreadLocal key = e.get();
                if (key != null) {
                    Object value = key.childValue(e.value);
                    Entry c = new Entry(key, value);
                    int h = key.threadLocalHashCode & (len - 1);
                    while (table[h] != null)
                        h = nextIndex(h, len);
                    table[h] = c;
                    size++;
                }
            }
        }
    }

     /**
     * 根据key获取到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);
    }

    /**
     * 当key没有在hash槽中出现的时候,需要根据这个方法进行获取
     */
    private Entry getEntryAfterMiss(ThreadLocal key, int i, Entry e) {
        Entry[] tab = table;
        int len = tab.length;
        while (e != null) {
            ThreadLocal k = e.get();
            if (k == key)
                return e;
            if (k == null)
                expungeStaleEntry(i);
            else
                i = nextIndex(i, len);
            e = tab[i];
        }
        return null;
    }

    /**
     * 设置key的值为value
     */
    private void set(ThreadLocal key, Object value) {

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

    /**
     * 移除key
     */
    private void remove(ThreadLocal key) {
        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)]) {
            if (e.get() == key) {
                e.clear();
                expungeStaleEntry(i);
                return;
            }
        }
    }

    /**
     * Replace a stale entry encountered during a set operation
     * with an entry for the specified key.  The value passed in
     * the value parameter is stored in the entry, whether or not
     * an entry already exists for the specified key.
     *
     * As a side effect, this method expunges all stale entries in the
     * "run" containing the stale entry.  (A run is a sequence of entries
     * between two null slots.)
     */
    private void replaceStaleEntry(ThreadLocal key, Object value,
                                   int staleSlot) {
        Entry[] tab = table;
        int len = tab.length;
        Entry e;

        // Back up to check for prior stale entry in current run.
        // We clean out whole runs at a time to avoid continual
        // incremental rehashing due to garbage collector freeing
        // up refs in bunches (i.e., whenever the collector runs).
        int slotToExpunge = staleSlot;
        for (int i = prevIndex(staleSlot, len);
             (e = tab[i]) != null;
             i = prevIndex(i, len))
            if (e.get() == null)
                slotToExpunge = i;

        // Find either the key or trailing null slot of run, whichever
        // occurs first
        for (int i = nextIndex(staleSlot, len);
             (e = tab[i]) != null;
             i = nextIndex(i, len)) {
            ThreadLocal k = e.get();

            // If we find key, then we need to swap it
            // with the stale entry to maintain hash table order.
            // The newly stale slot, or any other stale slot
            // encountered above it, can then be sent to expungeStaleEntry
            // to remove or rehash all of the other entries in run.
            if (k == key) {
                e.value = value;

                tab[i] = tab[staleSlot];
                tab[staleSlot] = e;

                // Start expunge at preceding stale entry if it exists
                if (slotToExpunge == staleSlot)
                    slotToExpunge = i;
                cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
                return;
            }

            // If we didn't find stale entry on backward scan, the
            // first stale entry seen while scanning for key is the
            // first still present in the run.
            if (k == null && slotToExpunge == staleSlot)
                slotToExpunge = i;
        }

        // If key not found, put new entry in stale slot
        tab[staleSlot].value = null;
        tab[staleSlot] = new Entry(key, value);

        // If there are any other stale entries in run, expunge them
        if (slotToExpunge != staleSlot)
            cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
    }

    /**
     * Expunge a stale entry by rehashing any possibly colliding entries
     * lying between staleSlot and the next null slot.  This also expunges
     * any other stale entries encountered before the trailing null.  See
     * Knuth, Section 6.4
     *
     * @param staleSlot index of slot known to have null key
     * @return the index of the next null slot after staleSlot
     * (all between staleSlot and this slot will have been checked
     * for expunging).
     */
    private int expungeStaleEntry(int staleSlot) {
        Entry[] tab = table;
        int len = tab.length;

        // expunge entry at staleSlot
        tab[staleSlot].value = null;
        tab[staleSlot] = null;
        size--;
        // Rehash until we encounter null
        Entry e;
        int i;
        for (i = nextIndex(staleSlot, len);(e = tab[i]) != null;i = nextIndex(i, len)) {
            ThreadLocal k = e.get();
            if (k == null) {
                e.value = null;
                tab[i] = null;
                size--;
            } else {
                int h = k.threadLocalHashCode & (len - 1);
                if (h != i) {
                    tab[i] = null;
                    while (tab[h] != null)
                        h = nextIndex(h, len);
                    tab[h] = e;
                }
            }
        }
        return i;
    }



    /**
     * Re-pack and/or re-size the table. First scan the entire
     * table removing stale entries. If this doesn't sufficiently
     * shrink the size of the table, double the table size.
     */
    private void rehash() {
        expungeStaleEntries();
        // Use lower threshold for doubling to avoid hysteresis
        if (size >= threshold - threshold / 4)
            resize();
    }

    /**
     * Double the capacity of the table.
     */
    private void resize() {
        Entry[] oldTab = table;
        int oldLen = oldTab.length;
        int newLen = oldLen * 2;
        Entry[] newTab = new Entry[newLen];
        int count = 0;
        for (int j = 0; j < oldLen; ++j) {
            Entry e = oldTab[j];
            if (e != null) {
                ThreadLocal k = e.get();
                if (k == null) {
                    e.value = null; // Help the GC
                } else {
                    int h = k.threadLocalHashCode & (newLen - 1);
                    while (newTab[h] != null)
                        h = nextIndex(h, newLen);
                    newTab[h] = e;
                    count++;
                }
            }
        }
        setThreshold(newLen);
        size = count;
        table = newTab;
    }
}

1、set方法

public void set(T value) {  
   Thread t = Thread.currentThread();//1.首先获取当前线程对象  
       ThreadLocalMap map = getMap(t);//2.获取该线程对象的ThreadLocalMap  
       if (map != null)  
           map.set(this, value);//如果map不为空,执行set操作,以当前threadLocal对象为key,实际存储对象为value进行set操作  
       else  
           createMap(t, value);//如果map为空,则为该线程创建ThreadLocalMap  
   } 
ThreadLocalMap getMap(Thread t) {
    return t.threadLocals;
}

//threadLocals是在Thread类中定义的变量
ThreadLocal.ThreadLocalMap threadLocals = null;

// 若该线程没有ThreadLocalMap对象,需要进行创建
void createMap(Thread t, T firstValue) {
    t.threadLocals = new ThreadLocalMap(this, firstValue);
}

2、get方法

获取到当前线程的ThreadLocalMap对象,然后获取到该对象的值

如果没有的话,就创建该ThreadLocalMap

public T get() {
    Thread t = Thread.currentThread(); // 获取到当前线程
    ThreadLocalMap map = getMap(t); // 获取到当前线程的ThreadLocalMap对象
    // 如果不为空就尝试获取值,为空就调用setInitialValue来进行创建初始值的ThreadLocalMap对象
    if (map != null) {
        ThreadLocalMap.Entry e = map.getEntry(this);
        if (e != null)
            return (T)e.value;
    }
    return setInitialValue();
}

private T setInitialValue() {
     T value = initialValue();
     Thread t = Thread.currentThread();
     ThreadLocalMap map = getMap(t);
     if (map != null)
        map.set(this, value);
     else
        createMap(t, value);
     return value;
}

4、ReentrantLock的底层

ReentrantLock的源码:

(1)ReentrantLock是Lock的子类,支持序列化

(2)ReentrantLock里面有一抽象的静态内部类Sync(继承了AQS),并且有一类型为该类Sync的成员变量

​ 有两个Sync的子类,表示公平锁和非公平锁

(3)ReentrantLock里面的方法都是通过sync调用其里面的方法进行实现的

(4)ReentrantLock里面是默认使用非公平锁的,如果要使用公平锁,则在lock定义的时候需要传入参数true

state初始化为0,表示未锁定状态。A线程lock时,会调用tryAcquire独占该锁并将state+acquires(一般acquires就是1)

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;

    private final Sync sync;

    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;

        /**
        * 加锁
        */
        abstract void lock();

        /**
        * 非公平锁尝试获取锁
        */
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            // 0 表示未锁定状态
            int c = getState();
            if (c == 0) {
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
        /**
        * 尝试释放资源,成功则返回true,失败则返回false。
        */
        protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
        /**
        * 该线程是否正在独占资源。只有用到condition才需要去实现它
        */
        protected final boolean isHeldExclusively() {
            return getExclusiveOwnerThread() == Thread.currentThread();
        }

        final ConditionObject newCondition() {
            return new ConditionObject();
        }

        final Thread getOwner() {
            return getState() == 0 ? null : getExclusiveOwnerThread();
        }

        final int getHoldCount() {
            return isHeldExclusively() ? getState() : 0;
        }

        final boolean isLocked() {
            return getState() != 0;
        }

        private void readObject(java.io.ObjectInputStream s)
            throws java.io.IOException, ClassNotFoundException {
            s.defaultReadObject();
            setState(0); // reset to unlocked state
        }
    }

    public ReentrantLock() {
        sync = new NonfairSync();
    }

    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

    public void lock() {
        sync.lock();
    }

    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    public boolean tryLock() {
        return sync.nonfairTryAcquire(1);
    }

    public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }

    public void unlock() {
        sync.release(1);
    }

    public Condition newCondition() {
        return sync.newCondition();
    }


    public int getHoldCount() {
        return sync.getHoldCount();
    }


    public boolean isHeldByCurrentThread() {
        return sync.isHeldExclusively();
    }


    public boolean isLocked() {
        return sync.isLocked();
    }


    public final boolean isFair() {
        return sync instanceof FairSync;
    }


    protected Thread getOwner() {
        return sync.getOwner();
    }


    public final boolean hasQueuedThreads() {
        return sync.hasQueuedThreads();
    }


    public final boolean hasQueuedThread(Thread thread) {
        return sync.isQueued(thread);
    }


    public final int getQueueLength() {
        return sync.getQueueLength();
    }


    protected Collection<Thread> getQueuedThreads() {
        return sync.getQueuedThreads();
    }

    public boolean hasWaiters(Condition condition) {
        if (condition == null)
            throw new NullPointerException();
        if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
            throw new IllegalArgumentException("not owner");
        return sync.hasWaiters((AbstractQueuedSynchronizer.ConditionObject)condition);
    }


    public int getWaitQueueLength(Condition condition) {
        if (condition == null)
            throw new NullPointerException();
        if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
            throw new IllegalArgumentException("not owner");
        return sync.getWaitQueueLength((AbstractQueuedSynchronizer.ConditionObject)condition);
    }


    protected Collection<Thread> getWaitingThreads(Condition condition) {
        if (condition == null)
            throw new NullPointerException();
        if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
            throw new IllegalArgumentException("not owner");
        return sync.getWaitingThreads((AbstractQueuedSynchronizer.ConditionObject)condition);
    }


    public String toString() {
        Thread o = sync.getOwner();
        return super.toString() + ((o == null) ?
                                   "[Unlocked]" :
                                   "[Locked by thread " + o.getName() + "]");
    }
}

重入锁与非重入锁

ReentrantLock是一个可重入锁,尝试获取锁tryAcquire方法是不会引起阻塞的,lock方法是会引起阻塞的

先列举出其源码:

重入锁尝试获取锁的源码如下:

/**
    * 尝试获取资源,成功则返回true,失败则返回false。
     */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        //首先判断当前是否已经有线程获取到锁 , 0表示没有获取到锁
        if (c == 0) {
            if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current); // 设置当前线程为独占该锁的线程
                return true;
            }
            // 若已经有线程获取到锁了,则判断这个线程是不是就是当前的线程(可重入的原因)
        }else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
  • 可重入锁尝试获取资源的方法逻辑是,首先判断当前是否已经有线程获取到锁
  • 若当前锁没有被占用,则当前线程可以进行占用到这个锁,变为独占锁
  • 若当前锁被占用了,这个时候要比较这个占用的锁的线程是不是当前线程,如果是,也相当于获取到锁的。着也是与非重入锁差别的一点

若改为非重入锁,尝试获取锁的源码如下:

/**
    * 尝试获取资源,成功则返回true,失败则返回false。
     */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        return false;
    }
  • 不可重入锁尝试获取资源的方法逻辑是,首先判断当前是否已经有线程获取到锁
  • 若当前锁没有被占用,则当前线程可以进行占用到这个锁,变为独占锁
  • 若当前锁被占用了,这个时候就直接返回获取锁失败false(它不会去判断占用这个锁的线程是不是就是当前线程)

公平锁与非公平锁

ReentrantLock里面是默认使用非公平锁的,如果要使用公平锁,则在lock定义的时候需要传入参数true

state初始化为0,表示未锁定状态。A线程lock时,会调用tryAcquire独占该锁并将state+acquires

(1)公平锁与非公平锁的区别:

体现在获取锁(调用lock方法)的时候,非公平锁是先进行CAS操作,进行判断是否可以直接获取到锁,则公平锁就直接进行获取锁的操作

// 非公平锁
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;
     /**
     * 先进行CAS判断,是否可以获取到锁,如果可以获取到锁,就将当前独占的锁设置为当前进程
     */
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
    // nonfairTryAcquire查看上面Sync中的
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

// 公平锁
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;
    /**
    * state 为 0 表示未锁定状态,加锁,就给state加1
    */
    final void lock() {
        acquire(1);
    }

    /**
    * 尝试获取资源,成功则返回true,失败则返回false。
     */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}

// 获取到锁
 public final void acquire(int arg) {
        if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
}

AQS中的CAS操作,通过预期值与内存的值进行比较,若相同,则进行更新

protected final boolean compareAndSetState(int expect, int update) {
    // unsafe一般用于原子Atmoic类中,底层相关的东西
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

全部评论

(0) 回帖
加载中...
话题 回帖

推荐话题

相关热帖

近期热帖

近期精华帖

热门推荐