网站建设网银,wordpress模块怎么设置在最上层,wordpress提示框美化,asp网站源码 生成静态在实习过程中使用阻塞队列对while sleep 轮询机制进行了改造#xff0c;提升了发送接收的效率#xff0c;这里做一点点总结。
自从Java 1.5之后#xff0c;在java.util.concurrent包下提供了若干个阻塞队列#xff0c;BlcokingQueue继承了Queue接口#xff0c;是线程安全…在实习过程中使用阻塞队列对while sleep 轮询机制进行了改造提升了发送接收的效率这里做一点点总结。
自从Java 1.5之后在java.util.concurrent包下提供了若干个阻塞队列BlcokingQueue继承了Queue接口是线程安全的。阻塞功能使得生产者和消费者两端的能力得以平衡当阻塞队列是空时从队列中获取元素的操作将会被阻塞。当阻塞队列是满时往队列里添加元素的操作将会被阻塞。
等待唤醒机制 condition/await/singal 和阻塞队列的关系将在第四部分介绍。
比较和交换Compare And SwapCAS 将在文章第五部分介绍。
用过 Golang 的会发现这怎么这么像 channel之后应该会写写关于这俩的博客。 文章目录#1 阻塞队列的种类#2 常用方法队列主要方法阻塞队列主要方法#3 阻塞队列的原理 - 以 ArrayBlockingQueue 为例put() 方法take() 方法非阻塞队列ConcurrentLinkedQueue#4 等待唤醒机制 wait/notify/notifyAll、condition/await/singal轮询wait/notify/notifyAllcondition/await/singal#5 多线程安全 - CAS#1 阻塞队列的种类
ArrayBlockingQueue基于数组实现的一个阻塞队列在创建ArrayBlockingQueue对象时必须制定容量大小。并且可以指定公平性与非公平性默认情况下为非公平的即不保证等待时间最长的队列最优先能够访问队列。LinkedBlockingQueue基于链表实现的一个阻塞队列在创建LinkedBlockingQueue对象时如果不指定容量大小则默认大小为Integer.MAX_VALUE可被看作无界队列。PriorityBlockingQueue以上2种队列都是先进先出队列而PriorityBlockingQueue 会按照元素的优先级对元素进行排序每次出队的元素都是优先级最高的元素。此阻塞队列为无界阻塞队列即容量没有上限。DelayQueue基于PriorityQueue一种延时阻塞队列也有比较排序能力。DelayQueue中的元素只有当其指定的延迟时间到了才能够从队列中获取到该元素。DelayQueue也是一个无界队列因此往队列中插入数据的操作生产者永远不会被阻塞而只有获取数据的操作消费者才会被阻塞。SynchronousQueue它的容量为0不是1所以没有地方来暂存元素导致每次取数据都要先阻塞直到有数据放入同理每次放数据的时候也会阻塞直到有消费者来取。它不需要去持有元素它做的就是直接传递。LinkedTransferQueue是一个由链表结构组成的无界阻塞TransferQueue队列多了tryTransfer可以设置等待时间和transfer等待有消费者后这个方法才会执行下去方法。可以算是 LinkedBolckingQueue 和 SynchronousQueue 和合体。基于CAS无锁算法实现性能理论高一点。
Queue接口下面还有一种非阻塞队列ConcurrentLickedQueue不会使线程阻塞而是使用CAS非阻塞算法不停重试来实现线程安全的。 #2 常用方法
队列主要方法
对于非阻塞队列一般情况下建议使用offer、poll和peek三个方法因为使用offer、poll和peek三个方法可以通过返回值判断操作成功与否而使用add和remove方法却不能达到这样的效果。注意非阻塞队列中的方法都没有进行同步措施。
add(E e):将元素e插入到队列末尾如果插入成功则返回true如果插入失败即队列已满则会抛出异常remove()移除队首元素若移除成功则返回true如果移除失败队列为空则会抛出异常element()返回队列的头节点但是不会删除这个元素。当队列为空时同样会报NoSuchElementException的错误offer(E e)将元素e插入到队列末尾如果插入成功则返回true如果插入失败即队列已满则返回falsepoll()移除并获取队首元素若成功则返回队首元素否则返回nullpeek()获取队首元素若成功则返回队首元素否则返回null
阻塞队列主要方法
上面列举的5个方法在阻塞队列中都存在但是要注意这5个方法在阻塞队列中都进行了同步措施。
put(E e)阻塞方法向队尾存入元素如果队列满则等待take()阻塞方法从队首取元素如果队列为空则等待offer(E e,long timeout, TimeUnit unit)向队尾存入元素如果队列满则等待一定的时间当时间期限达到时如果还没有插入成功则返回false否则返回truepoll(long timeout, TimeUnit unit)从队首取元素如果队列空则等待一定的时间当时间期限达到时如果取到则返回null否则返回取得的元素
操作类型抛出异常阻塞线程有返回值超时返回存入addputofferoffer(E e, long timeout, TimeUnit unit)取出removetakepollpoll(long timeout, TimeUnit unit)
#3 阻塞队列的原理 - 以 ArrayBlockingQueue 为例
ArrayBlockingQueue中用来存储元素的实际上是一个数组takeIndex和putIndex分别表示队首元素和队尾元素的下标count表示队列中元素的个数。lock是一个可重入锁notEmpty和notFull是等待条件。
读操作和写操作都需要先获取到 ReentrantLock 独占锁才能进行下一步操作。进行读操作时如果队列为空线程就会进入到读线程专属的notEmpty的 Condition 的队列中去排队等待写线程写入新的元素同理队列已满这个时候写操作的线程进入到写线程专属的notFull队列中去排队等待读线程将队列元素移除并腾出空间。
public class ArrayBlockingQueueE extends AbstractQueueE
implements BlockingQueueE, java.io.Serializable {private static final long serialVersionUID -817911632652898426L;/** 用于存放元素的数组 The queued items */private final E[] items;/** 下一次读取的位置 items index for next take, poll or remove */private int takeIndex;/** 下一次写入的位置 items index for next put, offer, or add. */private int putIndex;/** 队列中元素的数量 Number of items in the queue */private int count;/** Concurrency control uses the classic two-condition algorithm* found in any textbook.*//** Main lock guarding all access */private final ReentrantLock lock;/** Condition for waiting takes */private final Condition notEmpty;/** Condition for waiting puts */private final Condition notFull;
}put() 方法
在put方法中首先用checkNotNull方法检查插入的元素是不是null如果不是 null 会用Reentrantlock上锁并且使用的上锁方法是lock.lockInterruptibly()。这个方法在获取锁的同时是可以响应中断的这也正是我们的阻塞队列调用put方法时在尝试获取锁但还没拿到锁的期间可以响应中断的底层原因。
紧接着在while循环中它会检查当前队列是不是已经满了也就是count的长度是否等于数组长度。如果等于代表已经满了于是我们便会进行等待直到有空余的时候我们才会执行下一步操作调用enqueue方法让元素进入队列最后用unlock方法解锁。
public void put(E e) throws InterruptedException {//判断添加元素e是否为空为空会抛出异常checkNotNull(e);//获取锁final ReentrantLock lock this.lock;//加锁lock.lockInterruptibly();try {//队列已满线程等待while (count items.length)notFull.await();//入队方法 signal()藏在enqueue(e)方法里了enqueue(e);} finally {//释放锁lock.unlock();}}private void enqueue(E x) {// assert lock.getHoldCount() 1;// assert items[putIndex] null;final Object[] items this.items;//插入数据items[putIndex] x;if (putIndex items.length)putIndex 0;count;//通知消费者线程当前队列中有数据可供消费notEmpty.signal();
}
take() 方法
跟put方法实现很类似只不过put方法等待的是notFull信号而take方法等待的是notEmpty信号。
//从队列中取出数据
public E take() throws InterruptedException {//获取锁final ReentrantLock lock this.lock;//加锁lock.lockInterruptibly();try {//当队列中无数据删除线程阻塞等待while (count 0)notEmpty.await();//有数据执行队头元素出队操作return dequeue();} finally {//释放锁lock.unlock();}}
非阻塞队列ConcurrentLinkedQueue
顺便看看使用CAS非阻塞算法不停重试来实现线程安全的这个队列。
ConcurrentLinkedQueue是一种非阻塞的线程安全队列与阻塞队列LinkedBlockingQueue相对应同样也是使用链表实现的FIFO队列但不同的是它没有使用任何锁机制而是用CAS来实现线程安全。
public boolean offer(E e) {// 检查e是不是null是的话抛出NullPointerException异常。checkNotNull(e);// 创建新的节点final NodeE newNode new NodeE(e);// 将“新的节点”添加到链表的末尾。for (NodeE t tail, p t;;) {NodeE q p.next;// 情况1q为空p是尾节点插入if (q null) {// CAS操作如果“p的下一个节点为null”(即p为尾节点)则设置p的下一个节点为newNode。// 如果该CAS操作成功的话则比较“p和t”(若p不等于t则设置newNode为新的尾节点)然后返回true。// 如果该CAS操作失败这意味着“其它线程对尾节点进行了修改”则重新循环。if (p.casNext(null, newNode)) {if (p ! t) // hop two nodes at a timecasTail(t, newNode); // Failure is OK.return true;}}// 情况2p和q相等else if (p q)p (t ! (t tail)) ? t : head;// 情况3其它elsep (p ! t t ! (t tail)) ? t : q;}
}非阻塞队列ConcurrentLinkedQueue主要流程就是在判空以后会进入一个大循环中p.casNext()方法这个方法正是利用了CAS来操作的这个死循环去配合CAS其实就是我们平时说的乐观锁的思想。
#4 等待唤醒机制 wait/notify/notifyAll、condition/await/singal
从上一部分可以发现阻塞队列 等待唤醒机制 队列例如 ArrayBlockingQueue 使用了 ReentrantLock Condition 和一个列表队列实现。这里就顺便说说 Java 中的阻塞唤醒机制。此部分参考这篇好文。
轮询
生产者和消费者线程各自使用while循环每隔片刻就去判断Queue的状态队列为空时生产者才可插入数据队列不为空时消费者才能取出数据否则一律sleep等待。
轮询最简单但是也最不优雅。
sleep 时不会释放线程拥有的锁。轮询的方式太耗费CPU资源如果线程过多比如几百上千个线程同时在那轮询会给CPU带来较大负担。无法保证原子性如果生产者的操作非原子性消费者极可能获取到脏数据。
wait/notify/notifyAll
等待唤醒机制底层通过维护线程队列的方式避免了过多线程同时自旋造成的CPU资源浪费颇有点“用空间换时间”的味道。
当一个生产者线程无法插入数据时就让它在队列里休眠阻塞此时生产者线程会释放CPU资源等到消费者抢到CPU执行权并取出数据后再由消费者唤醒生产者继续生产。
用synchronized保证原子性wait和notify实现等待唤醒
// 线程consumer
synchronize(obj){ obj.wait();//没东西了等待
}
// 线程producer
synchronize(obj){ obj.notify();//有东西了唤醒
}在synchronized机制下所有等待的线程都在同一个队列里而notify又恰巧是随机唤醒线程也就是说有可能生产者唤醒生产者。
一般建议是改用notifyAll把所有线程都唤醒然后大家一起参与执行权的竞争。
如果线程调用了对象的 wait()方法那么线程便会处于该对象的等待池中等待池中的线程不会去竞争该对象的锁。当有线程调用了对象的 notifyAll()方法唤醒所有 wait 线程或 notify()方法只随机唤醒一个 wait 线程被唤醒的的线程便会进入该对象的锁池中锁池中的线程会去竞争该对象锁。也就是说调用了notify后只要一个线程会由等待池进入锁池而notifyAll会将该对象等待池内的所有线程移动到锁池中等待锁竞争优先级高的线程竞争到对象锁的概率大假若某线程没有竞争到该对象锁它还会留在锁池中唯有线程再次调用 wait()方法它才会重新回到等待池中。而竞争到对象锁的线程则继续往下执行直到执行完了 synchronized 代码块它会释放掉该对象锁这时锁池中的线程会继续竞争该对象锁。
condition/await/singal
虽然 wait/notifyAll 能解决“己方唤醒己方最终导致全部线程阻塞”的乌龙事件但唤醒全部线程又不够精确会造成无谓的线程竞争实际只需要唤醒敌方线程即可。
作为改进版可以使用 ReentrantLock 的 Condition 替代 synchronized 的 wait/notify
// 显式锁相对地synchronized锁被称为隐式锁
private final ReentrantLock lock new ReentrantLock();
private final Condition producerCondition lock.newCondition();
private final Condition consumerCondition lock.newCondition();//生产
lock.lock();
producerCondition.await();
lock.unlock(); //消费
lock.lock();
consumerCondition.signal();
lock.unlock();你可以认为lock.newCondition() 创建了一个队列调用producerCondition.await()会把生产者线程放入生产者的等待队列中当消费者调用producerCondition.signal() 时会唤醒从生产者的等待队列中唤醒一个生产者线程出来工作。
也就是说ReentrantLock的Condition通过拆分线程等待队列让线程的等待唤醒更加精确了想唤醒哪一方就唤醒哪一方。
可以看到上面的伪代码和前面阻塞队列的put() 方法逻辑几乎一模一样。
#5 多线程安全 - CAS
补充一下这个概念。
在计算机科学中比较和交换Compare And SwapCAS 是用于实现多线程同步的原子指令。 它将内存位置的内容与给定值进行比较只有在相同的情况下将该内存位置的内容修改为新的给定值。
它包含 3 个参数 CAS(V,E,N)。V 表示要更新的变量(内存值)E 表示预期值(旧的)N 表示新值。当且仅当 V 值等 于 E 值时才会将 V 的值设为 N如果 V 值和 E 值不同则说明已经有其他线程做了更新则当前线程什么都不做。最后CAS 返回当前 V 的真实值。
CAS 操作是抱着乐观的态度进行的(乐观锁)它总是认为自己可以成功完成操作。当多个线程同时 使用 CAS 操作一个变量时只有一个会胜出并成功更新其余均会失败。失败的线程不会被挂起仅是被告知失败并且允许再次尝试当然也允许失败的线程放弃操作。基于这样的原理 CAS 操作即使没有锁也可以发现其他线程对当前线程的干扰并进行恰当的处理。 独占锁是一种悲观锁synchronized就是一种独占锁会导致其它所有需要锁的线程挂起等待持有锁的线程释放锁。而另一个更加有效的锁就是乐观锁。所谓乐观锁就是每次不加锁而是假设没有冲突而去完成某项操作如果因为冲突失败就重试直到成功为止。乐观锁用到的机制就是CASCompare and Swap。 自旋 就是不停的判断比较看能否将值交换 CAS的三大问题
ABA问题一个值原来是A变成了B又变成了A那么使用CAS进行检查时会发现它的值没有发生变化但是实际上却变化了。ABA问题的解决思路就是使用版本号每次变量更新的时候把版本号加一那么ABA 就会变成1A-2B3A。循环时间长开销大自旋CAS如果长时间不成功会给CPU带来非常大的执行开销。如果JVM能支持处理器提供的pause指令那么效率会有一定的提升。只能保证一个共享变量的原子操作对多个共享变量操作时循环CAS就无法保证操作的原子性这个时候就可以用锁。还有一个取巧的办法就是把多个共享变量合并成一个共享变量来操作。
