各种并发类的使用场景
wait notify
他们两个组合适用于比较简单的场景,由于 nofity 并不能执行唤醒的对象,就会导致如果唤醒了一个执行时机不正确的情况,所以被唤醒的线程还需要判断当前状态是否适合自己执行,如果不适合,就需要重新就行wait ,并通知其他线程, 在线程调度过程中会涉及到 线程的上下文切换,这也是一个比较消耗资源的操作,
他比较适合简单的生产消费模式
虽然说wait notify 能实现生产消费模式,但是写起来代码量还是很多的, 如果使用ArrayBlockingQueue 的put 与 take方法,他们组合本身就是一个生产消费模式,使用过程中的代码量非常少,
val queue=ArrayBlockingQueue<Int>(20)
thread {
for (i in 0..10000){
var i= queue.take()
Log.i("tian.shm","取出一个$i")
sleep(500)
}
}
thread {
for (i in 0..10000){
queue.put(i)
Log.i("tian.shm","添加一个")
sleep(250)
}
}
这里需要介绍一下BlockingQueue 的3组方法
add remove --> 不满足当前情况就会抛出异常
put take --> 阻塞队列
poll offer --> 返回执行结果,如果没执行成功返回false
所以在线程池的队列中,我们可以根据业务使用不同的策略来适配
ArrayBlockingQueue 的实现原理
ArrayBlockingQueue 是一个静态数组,他有2把锁, notEmptyLock notFullLock , put 与 take的实现方式如下
put
public void put(E e) throws InterruptedException {
Objects.requireNonNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
lock.unlock();
}
}
可以看到在put 的过程中是先拿到锁,如果满了就让 notFull 等待,直到有位置后,加入队列
而take方法的实现如下
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
就一个地方不一样,那就是如果现在数据没有为了,等待被唤醒
countdownlatch
一个非常好的计数条件, 他是基于 AQS 的 Shared 方案来实现的,即可以同时多个线程wait ,当计数器为0时,同时启用多个等待的线程, countdownlatch 比较好的使用场景就是 android 的启动任务优化,如果某一个任务的启动调试是前面的某些个任务,那么他的启动时机就是前面所有的任务都启动完成,在我的App 启动任务优化方案中使用的就是这个方法,大家可以看一下,我写了非常多的注释 https://github.com/tsm1991/StartUp
LinkedBlockingQueue
他是一个链表的阻塞队列,在线程池中使用它由于capacity 这个参数设置的是 Integer.MAX_VALUE 就代表着所有的任务都可以进入到队列中,并不会被抛弃,在 LinkedBlockingQueue 的设计当中,他的数据插入与 数据读取使用的是2把锁,2把锁互不干扰,这个理解起来也比较好理解,在插入的过程中不应读取,只会影响到后续的插入,反之亦然,但是在删除节点的过程中的他调用了fulllock ,也就是将2个锁都锁起来了,就代表着,如果想要做数据的删除操作时 就与 ArrayBlockingQueue 比较相似了,其他操作都不能做了,下面看一下他的 poll put remove
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;
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.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
由于我们介绍过 capacity 是 Integer.MAX_VALUE ,所以 notFull 与 notEmpty 的锁就不介绍了,其实他的实现原理与 ArrayBlockingQueue 相类似的,这里注意使用的是 putLock
再来看poll
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E x = null;
int c = -1;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
if (nanos <= 0L)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
看一看到他使用的是 takeLock ,与插入数据不是同一把锁,所以说他的插入与读取是互不干扰的
但是到了remove 又是另一种情况了
public boolean remove(Object o) {
if (o == null) return false;
fullyLock();
try {
for (Node<E> trail = head, p = trail.next;
p != null;
trail = p, p = p.next) {
if (o.equals(p.item)) {
unlink(p, trail);
return true;
}
}
return false;
} finally {
fullyUnlock();
}
}
void fullyLock() {
putLock.lock();
takeLock.lock();
}
在删除节点的过程中,使用的是fulllock,也就是2把锁都锁住了
LinkedBlockingDeque
他是一个链表组成的双向阻塞队列,由于他是双向链表,并且读写锁是分离的,能保证数据的有序性,常常用于视频流的编解码,相较于LinkedBlockingQueue 只能一边插入与读取,破坏了数据的顺序,
