JDK容器類List，Set，Queue源碼解讀
- 2019 年 10 月 7 日
- 筆記
List，Set，Queue都是繼承Collection介面的單列集合介面。List常用的實現主要有ArrayList，LinkedList，List中的數據是有序可重複的。Set常用的實現主要是HashSet，Set中的數據是無序不可重複的。Queue常用的實現主要有ArrayBlockingQueue，LinkedBlockingQueue，Queue是一個保持先進先出的順序隊列，不允許隨機訪問隊列中的元素。
ArrayList核心源碼解讀
ArrayList是一個底層用數組實現的集合，數組元素類型為Object類型，支援隨機訪問，元素有序且可以重複，它繼承於AbstractList，實現了List, RandomAccess, Cloneable, java.io.Serializable這些介面。
當通過 ArrayList() 構造一個是集合，它是構造了一個空數組，初始長度為0。當第1次添加元素時，會創建一個長度為10的數組，並將該元素賦值到數組的第一個位置，當添加的元素大於10的時候，數組會進行第一次擴容。擴容1.5倍，長度變為15。 ArrayList在遍歷的時候時不能修改的，即遍歷的時候不能增加或者刪除元素，否則會拋ConcurrentModificationException
ArrayList是執行緒不安全的。
ArrayList源碼中的主要欄位
// 默認數組的大小 private static final int DEFAULT_CAPACITY = 10; // 默認空數組 private static final Object[] EMPTY_ELEMENTDATA = {}; // 實際存放數據的數組 private transient Object[] elementData;
ArrayList源碼中的構造器
/** * Constructs an empty list with an initial capacity of ten. */ public ArrayList() { this.elementData = DEFAULTCAPACITY_EMPTY_ELEMENTDATA; } /** * Constructs an empty list with the specified initial capacity. * * @param initialCapacity the initial capacity of the list * @throws IllegalArgumentException if the specified initial capacity * is negative */ public ArrayList(int initialCapacity) { if (initialCapacity > 0) { this.elementData = new Object[initialCapacity]; //將自定義的容量大小當成初始化elementData的大小 } else if (initialCapacity == 0) { this.elementData = EMPTY_ELEMENTDATA; } else { throw new IllegalArgumentException("Illegal Capacity: "+ initialCapacity); } }
ArrayList源碼中的add方法
/** * Appends the specified element to the end of this list. * * @param e element to be appended to this list * @return <tt>true</tt> (as specified by {@link Collection#add}) */ public boolean add(E e) { ensureCapacityInternal(size + 1); //添加元素之前，首先要確定集合的大小 elementData[size++] = e; //在數據中正確的位置上放上元素e，並且size++ return true; } private void ensureCapacityInternal(int minCapacity) { ensureExplicitCapacity(calculateCapacity(elementData, minCapacity)); } private static int calculateCapacity(Object[] elementData, int minCapacity) { if (elementData == DEFAULTCAPACITY_EMPTY_ELEMENTDATA) { // 如果為空數組 return Math.max(DEFAULT_CAPACITY, minCapacity); // 返回默認的我數組長度 } return minCapacity; } private void ensureExplicitCapacity(int minCapacity) { modCount++; // 修改次數+1，相當於版本號 // overflow-conscious code if (minCapacity - elementData.length > 0) // 如果實際容量小於需要容量 grow(minCapacity); // 擴容 } /** * Increases the capacity to ensure that it can hold at least the * number of elements specified by the minimum capacity argument. * * @param minCapacity the desired minimum capacity */ private void grow(int minCapacity) { // overflow-conscious code int oldCapacity = elementData.length; // 拿到數組的當前長度 int newCapacity = oldCapacity + (oldCapacity >> 1); //新數組的長度等於原數組長度的1.5倍 if (newCapacity - minCapacity < 0) //當新數組長度仍然比minCapacity小，則為保證最小長度，新數組等於minCapacity newCapacity = minCapacity; if (newCapacity - MAX_ARRAY_SIZE > 0) //當得到的新數組長度比 MAX_ARRAY_SIZE大時，調用 hugeCapacity 處理大數組 newCapacity = hugeCapacity(minCapacity); // minCapacity is usually close to size, so this is a win: elementData = Arrays.copyOf(elementData, newCapacity); //調用 Arrays.copyOf將原數組拷貝到一個大小為newCapacity的新數組（拷貝引用） } private static int hugeCapacity(int minCapacity) { if (minCapacity < 0) // overflow throw new OutOfMemoryError(); return (minCapacity > MAX_ARRAY_SIZE) ? Integer.MAX_VALUE : MAX_ARRAY_SIZE; // 數組的最大長度為Integer.MAX_VALUE }
ArrayList源碼中的get方法
/** * Returns the element at the specified position in this list. * * @param index index of the element to return * @return the element at the specified position in this list * @throws IndexOutOfBoundsException {@inheritDoc} */ public E get(int index) { rangeCheck(index); //判斷索引合法性 return elementData(index); }
執行緒安全的ArrayList—CopyOnWriteArrayList
CopyOnWriteArrayList是基於寫時複製（copy-on-write）思想來實現的一個執行緒安全的ArrayList集合類。它實現了List介面，內部持有一個ReentrantLock來給寫上鎖，底層是用volatile transient聲明的數組array，它是讀寫分離的，寫入數據時會複製出一個新的數組並加上ReentrantLock鎖，完成寫入後將新數組賦值給當前array，而讀操作不需要獲得鎖，支援並發。
CopyOnWriteArrayList的寫時複製導致了數據並不是實時的，有一定的延遲性，同時由於數據的複製，當數據量非常大的時候會佔用很大的記憶體。 CopyOnWriteArrayList是適合讀多寫少的場景。
CopyOnWriteArrayList核心源碼解讀
// 存放數據的數組 private transient volatile Object[] array; /** * 添加數據方法 * Appends the specified element to the end of this list. * * @param e element to be appended to this list * @return {@code true} (as specified by {@link Collection#add}) */ public boolean add(E e) { final ReentrantLock lock = this.lock; lock.lock(); // 加鎖，保證數據安全 try { Object[] elements = getArray(); // 拿到數組 int len = elements.length; // 獲取數組長度 Object[] newElements = Arrays.copyOf(elements, len + 1); // 拷貝數組到新數組 newElements[len] = e; // 將元素賦值到新數組的末尾 setArray(newElements); //設置新數組為當前數組 return true; } finally { lock.unlock(); // 解鎖 } }
HashSet源碼解讀
HashSet是最常用的Set實現，它繼承了AbstractSet抽象類，實現了Set，Cloneable和java.io.Serializable介面。
HashSet中存儲的是無序不可重複的數據，他的底層的數據存儲是通過HashMap實現的，HashSet將數據存儲在HashMap的key中，將HashMap的value設為一個Object引用。
HashSet核心源碼解讀
// 實際存儲數據的HashMap private transient HashMap<E,Object> map; // HashMap的value引用 private static final Object PRESENT = new Object(); /** * Constructs a new, empty set; the backing <tt>HashMap</tt> instance has * default initial capacity (16) and load factor (0.75). */ public HashSet() { map = new HashMap<>(); //new一個 HashMap 對象出來，採用無參的 HashMap 構造函數，具有默認初始容量（16）和載入因子（0.75）。} /** * Adds the specified element to this set if it is not already present. * More formally, adds the specified element <tt>e</tt> to this set if * this set contains no element <tt>e2</tt> such that * <tt>(e==null ? e2==null : e.equals(e2))</tt>. * If this set already contains the element, the call leaves the set * unchanged and returns <tt>false</tt>. * * @param e element to be added to this set * @return <tt>true</tt> if this set did not already contain the specified * element */ public boolean add(E e) { return map.put(e, PRESENT)==null; }
執行緒安全的HashSet—CopyOnWriteArraySet
CopyOnWriteArraySet是一個執行緒安全的HashSet，它底層是通過CopyOnWriteArrayList實現的，它是通過在添加數據的時候如果數據不存在才進行添加來實現了數據的不可重複
CopyOnWriteArraySet 核心源碼解讀
// 實際存放數據 private final CopyOnWriteArrayList<E> al; /** * Adds the specified element to this set if it is not already present. * More formally, adds the specified element {@code e} to this set if * the set contains no element {@code e2} such that * <tt>(e==null ? e2==null : e.equals(e2))</tt>. * If this set already contains the element, the call leaves the set * unchanged and returns {@code false}. * * @param e element to be added to this set * @return {@code true} if this set did not already contain the specified * element */ public boolean add(E e) { return al.addIfAbsent(e); // 如果不存在則添加 }
Queue詳細分析
Queue是先入先出（FIFO）的一個隊列數據結構，可以分為阻塞隊列和非阻塞隊列，Queue介面與List、Set同一級別，都是繼承了Collection介面。
Queue API
ArrayBlockingQueue是數組實現的執行緒安全的有界的阻塞隊列。
// 存放數據的數組 final Object[] items; // 取數據索引 int takeIndex; // 放數據索引 int putIndex; // 數據量 int count; // 鎖final ReentrantLock lock; /** Condition for waiting takes */ private final Condition notEmpty; /** Condition for waiting puts */ private final Condition notFull; /** items index for next put, offer, or add */ int putIndex; /** * Inserts the specified element at the tail of this queue if it is * possible to do so immediately without exceeding the queue's capacity, * returning {@code true} upon success and {@code false} if this queue * is full. This method is generally preferable to method {@link #add}, * which can fail to insert an element only by throwing an exception. * * @throws NullPointerException if the specified element is null */ public boolean offer(E e) { // offer方法是非阻塞 checkNotNull(e); final ReentrantLock lock = this.lock; // offer的時候加鎖 lock.lock(); try { if (count == items.length) // 如果沒有空間， 返回false return false; else { enqueue(e); // 如果有空間，入隊列 return true; } } finally { lock.unlock(); } } /** * Inserts the specified element at the tail of this queue, waiting * for space to become available if the queue is full. * * @throws InterruptedException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock = this.lock; // 加鎖 lock.lockInterruptibly(); try { while (count == items.length) // queue的容量已滿 notFull.await(); // 阻塞 enqueue(e); } finally { lock.unlock(); } } public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); //隊列為空時，將使這個執行緒進入阻塞狀態,直到被其他執行緒喚醒時取出元素 return dequeue(); //消費對頭中的元素 } finally { lock.unlock(); } }
LinkedBlockingQueue源碼解讀
LinkedBlockingQueue底層是採用鏈表實現的一個阻塞的執行緒安全的隊列。
LinkedBlockingQueue構造的時候若沒有指定大小，則默認大小為Integer.MAX_VALUE，當然也可以在構造函數的參數中指定大小。 LinkedBlockingQueue中採用兩把鎖，取數據的時候加takeLock，放數據的時候加putLock。
// 容量 private final int capacity; // 元素數量 private final AtomicInteger count = new AtomicInteger(); // 鏈表頭 transient Node<E> head; // 鏈表尾 private transient Node<E> last; // take鎖 private final ReentrantLock takeLock = new ReentrantLock(); // 當隊列無元素時，take鎖會阻塞在notEmpty條件上，等待其它執行緒喚醒 private final Condition notEmpty = takeLock.newCondition(); // 放鎖private final ReentrantLock putLock = new ReentrantLock(); // 當隊列滿了時，put鎖會會阻塞在notFull上，等待其它執行緒喚醒 private final Condition notFull = putLock.newCondition(); /** * Creates a {@code LinkedBlockingQueue} with a capacity of * {@link Integer#MAX_VALUE}. */ public LinkedBlockingQueue() { this(Integer.MAX_VALUE); // 如果沒傳容量，就使用最大int值初始化其容量 } /** * Inserts the specified element at the tail of this queue, waiting if * necessary for space to become available. * * @throws InterruptedException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); // Note: convention in all put/take/etc is to preset local var // holding count negative to indicate failure unless set. int c = -1; Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; // 使用putLock鎖加鎖 final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { /* * Note that count is used in wait guard even though it is * not protected by lock. This works because count can * only decrease at this point (all other puts are shut * out by lock), and we (or some other waiting put) are * signalled if it ever changes from capacity. Similarly * for all other uses of count in other wait guards. */ while (count.get() == capacity) { // 如果隊列滿了，就阻塞在notFull條件上 notFull.await(); // 等待被其它執行緒喚醒 } enqueue(node); // 隊列不滿了，就入隊 c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); } public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; // 使用takeLock加鎖 takeLock.lockInterruptibly(); try { while (count.get() == 0) { // 如果隊列無元素，則阻塞在notEmpty條件上 notEmpty.await(); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x; }
ConcurrentLinkedQueue源碼解讀
ConcurrentLinkedQueue是執行緒安全的無界非阻塞隊列，其底層數據結構是使用單向鏈表實現，入隊和出隊操作是使用我們經常提到的CAS來保證執行緒安全的。
ConcurrentLinkedQueue不允許插入的元素為null。
private transient volatile Node<E> head; private transient volatile Node<E> tail; private static final sun.misc.Unsafe UNSAFE; private static final long headOffset; private static final long tailOffset; /** * Inserts the specified element at the tail of this queue. * As the queue is unbounded, this method will never return {@code false}. * * @return {@code true} (as specified by {@link Queue#offer}) * @throws NullPointerException if the specified element is null */ public boolean offer(E e) { checkNotNull(e); // 為null則拋出空指針異常 final Node<E> newNode = new Node<E>(e); for (Node<E> t = tail, p = t;;) { // 自旋 Node<E> q = p.next; if (q == null) { // 如果q==null說明p是尾節點，則執行插入 // p is last node if (p.casNext(null, newNode)) { // 使用CAS設置p節點的next節點 // Successful CAS is the linearization point // for e to become an element of this queue, // and for newNode to become "live". if (p != t) // hop two nodes at a time casTail(t, newNode); // Failure is OK. return true; } // Lost CAS race to another thread; re-read next } else if (p == q) // We have fallen off list. If tail is unchanged, it // will also be off-list, in which case we need to // jump to head, from which all live nodes are always // reachable. Else the new tail is a better bet. p = (t != (t = tail)) ? t : head; else // Check for tail updates after two hops. p = (p != t && t != (t = tail)) ? t : q; } }
