ThreadLocal底层数据结构
相关的CSDN博客:
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) 回帖