源码仓库Gitee · GitHub一.Semaphore1.概念一个计数信号量。在概念上信号量维持一组许可证。如果有必要每个acquire()都会阻塞直到许可证可用然后才能使用它。每个release()添加许可证潜在地释放阻塞获取方。但是没有使用实际的许可证对象;Semaphore只保留可用数量的计数并相应地执行。即一个Semaphore维护了一组permits【许可证】。每次调用acquire()方法都会阻塞直到获取到许可证。每次调用release()方法都会添加一个许可证也就是释放一个被阻塞的获取者。但是实际上并不存在这个许可证Semaphore仅仅是记录可用资源的数量并且做出对应的行为有资源就获取没有资源就阻塞。信号量通常用于限制线程数而不是访问某些物理或逻辑资源。线程池控制的是线程数量而信号量控制的是并发数量虽然说看起来一样但两者还是有区别的。信号量类似于锁机制信号量的调用当达到数量后线程还是存在的只是被挂起了而已。而线程池同时执行的线程数量是固定的超过了数量的只能等待。在获得项目之前每个线程必须从信号量获取许可证以确保某个项目可用。当线程完成该项目后它将返回到池中并将许可证返回到信号量允许另一个线程获取该项目。请注意当调用acquire()时不会保持同步锁定因为这将阻止某个项目返回到池中。信号量封装了限制对池的访问所需的同步与保持池本身一致性所需的任何同步分开。【即将限制对池的访问和对池中数据的操作所需要的锁分开】。信号量被初始化为一个并且被使用使得它只有至多一个允许可用可以用作互斥锁。这通常被称为二进制信号量因为它只有两个状态一个许可证可用或零个许可证可用。当以这种方式使用时二进制信号量具有属性与许多Lock实现不同“锁”可以由除所有者之外的线程释放因为信号量没有所有权概念。这在某些专门的上下文中是有用的例如死锁恢复。Semaphore(int permits) 创建一个 Semaphore与给定数量的许可证和非公平公平设置。 Semaphore(int permits, boolean fair) 创建一个 Semaphore与给定数量的许可证和给定的公平设置。此类的构造函数可选择接受公平参数。当设置为false时此类不会保证线程获取许可的顺序。特别是闯入是允许的也就是说一个线程调用acquire()可以提前已经等待线程分配的许可证-在等待线程队列的头部逻辑新的线程将自己【新线程将自己放在等待线程队列的最前面】。当公平设置为真时信号量保证调用acquire方法的线程被选择以按照它们调用这些方法的顺序获得许可先进先出; FIFO【FIFO的顺序是指是依据到达方法内部的执行点的时间并不是方法执行的时间。】。请注意FIFO排序必须适用于这些方法中的特定内部执行点。因此一个线程可以在另一个线程之前调用acquire但是在另一个线程之后到达排序点并且类似地从方法返回。另请注意未定义的tryAcquire方法不符合公平性设置但将采取任何可用的许可证。【不定时的tryAcquire()方法会任意选取可用的许可证。】【非公平锁可以插队获取运行公平锁按照线程顺序执行。】通常用于控制资源访问的信号量应该被公平地初始化以确保线程没有被访问资源【确保没有线程因为长时间获取不到许可证而饿死】。当使用信号量进行其他类型的同步控制时非正常排序的吞吐量优势往往超过公平性。2.用法线程可以通过acquire()方法获取到一个许可然后对共享资源进行操作如果许可集已分配完了那么线程将进入等待状态直到其他线程释放许可才有机会再获取许可线程释放一个许可通过release()方法完成许可将被归还给Semaphore。public static void main(String[] args) { ExecutorService service Executors.newCachedThreadPool(); final Semaphore sp new Semaphore(3); for (int i 0; i 7; i) { Runnable runnable () - { try { sp.acquire(); } catch (InterruptedException e1) { e1.printStackTrace(); } System.out.println(线程 Thread.currentThread().getName() 进入当前已有 (3 - sp.availablePermits()) 个并发); try { Thread.sleep((long) (Math.random() * 10000)); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(线程 Thread.currentThread().getName() 即将离开); sp.release(); //下面代码有时候执行不准确因为其没有和上面的代码合成原子单元 System.out.println(线程 Thread.currentThread().getName() 已离开当前已有 (3 - sp.availablePermits()) 个并发); }; service.execute(runnable); } } 结果 线程pool-1-thread-1进入当前已有1个并发 线程pool-1-thread-2进入当前已有2个并发 线程pool-1-thread-3进入当前已有3个并发 线程pool-1-thread-3即将离开 线程pool-1-thread-4进入当前已有3个并发 线程pool-1-thread-3已离开当前已有3个并发 线程pool-1-thread-1即将离开 线程pool-1-thread-1已离开当前已有2个并发 线程pool-1-thread-5进入当前已有3个并发 线程pool-1-thread-5即将离开 线程pool-1-thread-5已离开当前已有2个并发 线程pool-1-thread-6进入当前已有3个并发 线程pool-1-thread-4即将离开 线程pool-1-thread-4已离开当前已有2个并发 线程pool-1-thread-7进入当前已有3个并发 线程pool-1-thread-2即将离开 线程pool-1-thread-2已离开当前已有2个并发 线程pool-1-thread-7即将离开 线程pool-1-thread-7已离开当前已有1个并发 线程pool-1-thread-6即将离开 线程pool-1-thread-6已离开当前已有0个并发3.acquire解析acquire() 从该信号量获取许可证阻止直到可用或线程为 interrupted 。 void acquire(int permits) 从该信号量获取给定数量的许可证阻止直到所有可用否则线程为 interrupted 。public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1);//调用AQS的acquireSharedInterruptibly } /** * AQS的acquireSharedInterruptibly * Acquires in shared mode, aborting if interrupted. Implemented * by first checking interrupt status, then invoking at least once * {link #tryAcquireShared}, returning on success. Otherwise the * thread is queued, possibly repeatedly blocking and unblocking, * invoking {link #tryAcquireShared} until success or the thread * is interrupted. * 以共享模式获取如果中断被中止。 * 实现首先检查中断状态然后至少调用一次tryacquirered成功返回。 * 否则,线程排队可能会重复阻塞和取消阻塞 * 调用tryacquiremred直到成功或线程被打断了。 */ public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted())//中断抛出异常 throw new InterruptedException(); if (tryAcquireShared(arg) 0)//获取失败加入同步队列等待 doAcquireSharedInterruptibly(arg); } //由Semaphore的FairSync或NonfairSync实现共享模式下资源可以被多个线程通知占用tryAcquireShared返回int类型表示还有多少个资源可以同时被占用用于共享模式下传播唤醒。 protected int tryAcquireShared(int arg) { throw new UnsupportedOperationException(); } //以共享中断模式获取 private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node addWaiter(Node.SHARED);//创建当前线程的节点并且锁的模型是共享锁将其添加到AQS CLH队列的末尾 boolean failed true; try { for (;;) {//自旋 final Node p node.predecessor();//获得当前节点的前驱节点 if (p head) {//是头节点没有等待节点 int r tryAcquireShared(arg); if (r 0) {//获取成功当前节点设置为头节点并传播【传播指的是同步状态剩余的许可数值不为0通知后续结点继续获取同步状态】 setHeadAndPropagate(node, r); p.next null; // help GC failed false; return; } } //前继节点非head节点没资源获取将前继节点状态设置为SIGNAL通过park挂起node节点的线程 if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node);//结束该结点线程的请求 } }public void acquire(int permits) throws InterruptedException { if (permits 0) throw new IllegalArgumentException();//数量小于0抛出异常 sync.acquireSharedInterruptibly(permits);//调用AQS的acquireSharedInterruptibly }tryAcquireSharedstatic final class FairSync extends Sync {//公平锁获取 protected int tryAcquireShared(int acquires) { for (;;) {//自旋 //有前驱节点表示当前线程前面有阻塞线程当前线程获取失败先让前节点线程获取运行【比非公平锁获取多了判断前驱节点的操作】 if (hasQueuedPredecessors()) return -1; int available getState();//可获取的许可证数量 int remaining available - acquires;//剩下的许可证数量 //如果剩余数量小于0或更新剩余数量成功返回剩余数量 if (remaining 0 || compareAndSetState(available, remaining)) return remaining; } } } static final class NonfairSync extends Sync {//非公平锁获取 protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires);//调用Semaphore的内部类Sync的nonfairTryAcquireShared } } final int nonfairTryAcquireShared(int acquires) { for (;;) {//自旋 int available getState(); int remaining available - acquires; if (remaining 0 || compareAndSetState(available, remaining)) return remaining; } }4.release解析public void release() { sync.releaseShared(1);//调用AQS的releaseShared } public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {//释放同步状态成功 doReleaseShared();//唤醒同步队列中后继结点的线程 return true; } return false; } protected boolean tryReleaseShared(int arg) {//由Semaphore的Sync实现 throw new UnsupportedOperationException(); } private void doReleaseShared() { /* * Ensure that a release propagates, even if there are other * in-progress acquires/releases. This proceeds in the usual * way of trying to unparkSuccessor of head if it needs * signal. But if it does not, status is set to PROPAGATE to * ensure that upon release, propagation continues. * Additionally, we must loop in case a new node is added * while we are doing this. Also, unlike other uses of * unparkSuccessor, we need to know if CAS to reset status * fails, if so rechecking. * 保证释放动作(向同步等待队列尾部)传递即使没有其他正在进行的 * 请求或释放动作。如果头节点的后继节点需要唤醒那么执行唤醒 * 动作如果不需要将头结点的等待状态设置为PROPAGATE保证 * 唤醒传递。另外为了防止过程中有新节点进入(队列)这里必 * 需做循环所以和其他unparkSuccessor方法使用方式不一样 * 的是如果(头结点)等待状态设置失败重新检测。 */ for (;;) { Node h head; if (h ! null h ! tail) { int ws h.waitStatus;//头节点状态 // 如果头节点对应的线程是SIGNAL状态则意味着头 //结点的后继结点所对应的线程需要被unpark-唤醒。 if (ws Node.SIGNAL) { // 修改头结点对应的线程状态设置为0。失败的话则继续循环。 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases // 唤醒头结点h的后继结点所对应的线程 unparkSuccessor(h); } else if (ws 0 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } // 如果头结点发生变化则继续循环。否则退出循环。 if (h head) // loop if head changed break; } } //唤醒传入结点的后继结点对应的线程 private void unparkSuccessor(Node node) { int ws node.waitStatus; if (ws 0) compareAndSetWaitStatus(node, ws, 0); //拿到后继结点 Node s node.next; if (s null || s.waitStatus 0) { s null; for (Node t tail; t ! null t ! node; t t.prev) if (t.waitStatus 0) s t; } if (s ! null) //唤醒该线程 LockSupport.unpark(s.thread); } protected final boolean tryReleaseShared(int releases) { for (;;) {//自旋 int current getState();//获取当前同步状态 int next current releases;//状态1 if (next current) // overflow throw new Error(Maximum permit count exceeded); if (compareAndSetState(current, next))//更新状态成功返回true return true; } }二.Exchanger1.概念线程可以在成对内配对和交换元素的同步点。每个线程在输入exchange方法时提供一些对象与合作者线程匹配并在返回时接收其合作伙伴的对象。交换器可以被视为一个的双向形式SynchronousQueue。交换器在诸如遗传算法和管道设计的应用中可能是有用的。一个用于两个工作线程之间交换数据的封装工具类简单说就是一个线程在完成一定的事务后想与另一个线程交换数据则第一个先拿出数据的线程会一直等待第二个线程直到第二个线程拿着数据到来时才能彼此交换对应数据。2.用法ExchangerV 泛型类型其中 V 表示可交换的数据类型V exchange(V v)等待另一个线程到达此交换点除非当前线程被中断然后将给定的对象传送给该线程并接收该线程的对象。V exchange(V v, long timeout, TimeUnit unit)等待另一个线程到达此交换点除非当前线程被中断或超出了指定的等待时间然后将给定的对象传送给该线程并接收该线程的对象。ExchangerInteger exchanger new Exchanger(); ExecutorService executor Executors.newCachedThreadPool(); Runnable run () -{ try { int num new Random().nextInt(10); System.out.println(Thread.currentThread().getName()开始交换数据:num); num exchanger.exchange(num);//交换数据并得到交换后的数据 System.out.println(Thread.currentThread().getName()交换数据结束后的数据num); } catch (InterruptedException e) { e.printStackTrace(); } }; executor.execute(run); executor.execute(run); executor.shutdown(); 结果 pool-1-thread-2开始交换数据:9 pool-1-thread-1开始交换数据:8 pool-1-thread-2交换数据结束后的数据8 pool-1-thread-1交换数据结束后的数据93.exchange源码解析 参考下面的博客写的很详细https://blog.csdn.net/qq_31865983/article/details/105620881https://www.xuebuyuan.com/2736097.html