内容简介：多个条件队列以实现更好的并发性。每个锁使用单独的条件队列的优点。以下是在无界队列之上的有界BlockingQueue的两个经典实现。
多个条件队列以实现更好的并发性。每个锁使用单独的条件队列的优点。
- 它避免了虚假的唤醒和上下文切换。例如，如果您使用notifyAll进行传统等待，则最终会唤醒正在等待不同条件的线程。
- 当您在单独的条件队列上等待时，您可以使用signal 而不是signalAll来进一步提高性能。
以下是在无界队列之上的有界BlockingQueue的两个经典实现。
每个锁具有单独的等待集
<b>public</b> <b>class</b> BlockingQueue<T> {
<b>private</b> <b>final</b> Queue<T> queue;
<b>private</b> <b>final</b> Lock lockObj = <b>new</b> ReentrantLock();
<b>private</b> <b>final</b> Condition empty = lockObj.newCondition();
<b>private</b> <b>final</b> Condition full = lockObj.newCondition();
<b>private</b> <b>int</b> maxLength;
<b>private</b> <b>int</b> currentSize = 0;
<b>public</b> BlockingQueue(<b>int</b> maxLength) {
<b>this</b>.queue = <b>new</b> ArrayDeque<T>();
<b>this</b>.maxLength = maxLength;
}
<b>public</b> <b>void</b> offer(T elem) throws InterruptedException {
lockObj.lock();
<b>try</b> {
<b>while</b> (currentSize == maxLength) {
full.await();
}
queue.offer(elem);
currentSize++;
empty.signal();
} <b>finally</b> {
lockObj.unlock();
}
}
<b>public</b> T poll() throws InterruptedException {
lockObj.lock();
<b>try</b> {
<b>while</b> (currentSize == 0) {
empty.await();
}
T elem = queue.poll();
currentSize--;
full.signal();
<b>return</b> elem;
} <b>finally</b> {
lockObj.unlock();
}
}
}
使用JMH测试吞吐量：
Benchmark Mode Cnt Score Error Units BenchmarkBlockingDequeu.testProduceAndConsume thrpt 25 12500542.933 ± 374127.076 ops/s
旧的方式（单锁和等待）
<b>public</b> <b>class</b> BlockingQueueWait<T> {
<b>private</b> <b>final</b> Queue<T> queue;
<b>private</b> <b>final</b> Object lockObj = <b>new</b> Object();
<b>private</b> <b>int</b> maxLength;
<b>private</b> <b>int</b> currentSize = 0;
<b>public</b> BlockingQueueWait(<b>int</b> maxLength) {
<b>this</b>.queue = <b>new</b> ArrayDeque<T>();
<b>this</b>.maxLength = maxLength;
}
<b>public</b> <b>void</b> offer(T elem) throws InterruptedException {
<b>synchronized</b> (lockObj) {
<b>while</b> (currentSize == maxLength) {
lockObj.wait();
}
queue.offer(elem);
currentSize++;
lockObj.notifyAll();
}
}
<b>public</b> T poll() throws InterruptedException {
<b>synchronized</b> (lockObj) {
<b>while</b> (currentSize == 0) {
lockObj.wait();
}
T elem = queue.poll();
currentSize--;
lockObj.notifyAll();
<b>return</b> elem;
}
}
}
使用JMH测试吞吐量：
Benchmark Mode Cnt Score Error Units BenchMarkBlockingWait.testProduceAndConsume thrpt 25 2702842.067 ± 24534.073 ops/s
如果你仔细看看上面的实现，ops /s的差异是巨大的，其中大部分是由虚假的唤醒引起的，并且没有使用显式条件队列和每个锁的等待集，你最终会浪费宝贵的cpu周期。因此，如果您正在编写并发库实现，请记住您有更好的并发支持，并且您可以为每个锁创建多个条件队列以避免虚假唤醒。
以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持 码农网
猜你喜欢:
- Redis 5.0 Stream消息队列会每次唤醒所有同一个group内的消费者
- 语音唤醒实现
- RunLoop的唤醒探索
- H5唤醒App之scheme方案
- 每日一博 | java 通过网络唤醒实现远程开机
- rabbitmq实现延时队列（死信队列）
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
