链表 结构 解读 Java 并发队列 BlockingQueue(12)

(oldCap + 2) :

(oldCap >> 1));

// 这里有可能溢出

if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow

int minCap = oldCap + 1;

if (minCap < 0 || minCap > MAX_ARRAY_SIZE)

throw new OutOfMemoryError();

newCap = MAX_ARRAY_SIZE;

}

// 如果 queue != array，那么说明有其他线程给 queue 分配了其他的空间

if (newCap > oldCap && queue == array)

// 分配一个新的大数组

newArray = new Object[newCap];

} finally {

// 重置，也就是释放锁

allocationSpinLock = 0;

}

}

// 如果有其他的线程也在做扩容的操作

if (newArray == null) // back off if another thread is allocating

Thread.yield();

// 重新获取锁

lock.lock();

// 将原来数组中的元素复制到新分配的大数组中

if (newArray != null && queue == array) {

queue = newArray;

System.arraycopy(array, 0, newArray, 0, oldCap);

}

}

扩容方法对并发的控制也非常的巧妙，释放了原来的独占锁 lock，这样的话，扩容操作和读操作可以同时进行，提高吞吐量。

下面，我们来分析下写操作 put 方法和读操作 take 方法。

public void put(E e) {

// 直接调用 offer 方法，因为前面我们也说了，在这里，put 方法不会阻塞

offer(e);

}

public boolean offer(E e) {

if (e == null)

throw new NullPointerException();

final ReentrantLock lock = this.lock;

// 首先获取到独占锁

lock.lock();

int n, cap;

Object[] array;

// 如果当前队列中的元素个数 >= 数组的大小，那么需要扩容了

while ((n = size) >= (cap = (array = queue).length))

tryGrow(array, cap);

try {

Comparator<? super E> cmp = comparator;

// 节点添加到二叉堆中

if (cmp == null)

siftUpComparable(n, e, array);

else

siftUpUsingComparator(n, e, array, cmp);

// 更新 size

size = n + 1;

// 唤醒等待的读线程

notEmpty.signal();

} finally {

lock.unlock();

}

return true;

}

对于二叉堆而言，插入一个节点是简单的，插入的节点如果比父节点小，交换它们，然后继续和父节点比较。

// 这个方法就是将数据 x 插入到数组 array 的位置 k 处，然后再调整树

本文来自电脑杂谈，转载请注明本文网址：
http://www.pc-fly.com/a/jisuanjixue/article-87792-12.html