1_基础知识_chapter05_基础构建模块_5_同步工具类

共同特点: 封装了一些状态, 用于决定执行同步工具类的线程继续执行还是等待

常见的有闭锁、信号量、栅栏等

闭锁

(1) 功能: 在闭锁到达结束状态之前, 门一直关闭没有任何线程可以通过;在闭锁到达结束状态后,门会打开, 允许所有线程通过,且永远保持打开

(2) 应用场景: 确保某些活动直到其他活动完成后才继续执行

1° 确保某个计算在其需要的所有资源都初始化以后才执行

2° 确保某个服务在其依赖的所有服务都启动后才启动

3° 等待某个操作的所有参与者都就绪再执行

(3) java.util.concurrent.CountDownLatch是一个闭锁的实现, 它提供了一个计数器(在构造函数中设定计数器初值); countDown()方法用于递减计数器,代表一个事件发生; await()方法用于等待计数器达到0,否则会一直阻塞到计数器为0, 或等待中的线程中断,或等待超时

(4) 示例

这个示例可以比较精确的统计所有线程的执行时间(并行), 它们将在计时后开始,在再次计时前停止

  public class TestHarness {

      public long timeTasks(int nThreads, final Runnable task) throws InterruptedException {

          final CountDownLatch startGate = new CountDownLatch(1);
          final CountDownLatch endGate = new CountDownLatch(nThreads);

          for (int i = 0; i < nThreads; i++) {

              Thread t = new Thread() {
                  
                  public void run() {
                      try {
                          startGate.await();
                          try {
                              task.run();
                          } finally {
                              endGate.countDown();
                          }
                      } catch (InterruptedException ignored) {
                      }
                  }
              };
              t.start();
          }

          long start = System.nanoTime();
  
          startGate.countDown();
          endGate.await();

          long end = System.nanoTime();
  
          return end - start;
      }
  }

信号量

(1) 功能: 控制同时访问某个特定资源的操作数量, 或者同时执行某个指定操作的数量

(2) 应用场景

1° 实现资源池(例如数据库连接池), 当池为空时阻塞, 直到非空时解除阻塞

2° 将任意一种容器, 变成有界阻塞容器

3° 当信号量为二元信号量时, 变成了互斥体mutex用于加锁

(3) java.util.concurrent.Semaphore是信号量的实现, 一个Semaphore对象管理着一组虚拟的许可, 许可的初始数量由构造函数指定; 当没有剩余许可时, acquire()将阻塞直到获得许可; release()将返回一个许可给Semaphore对象

(4) 示例

  public class BoundedHashSet<T> {

      private final Set<T> set;
      private final Semaphore sem;

      public BoundedHashSet(int bound) {

          this.set = Collections.synchronizedSet(new HashSet<T>());
          sem = new Semaphore(bound);
      }

      public boolean add(T o) throws InterruptedException {

          sem.acquire();
          boolean wasAdded = false;
  
          try {
              wasAdded = set.add(o);
              return wasAdded;
          } finally {
              if (!wasAdded) {
                  sem.release();
              }
          }
      }

      public boolean remove(Object o) {
  
          boolean wasRemoved = set.remove(o);
  
          if (wasRemoved) {
              sem.release();
          }
  
          return wasRemoved;
      }
  }

栅栏

(1) 功能: 所有线程必须同时到达栅栏, 才能继续执行

(2) 和闭锁的区别: 闭锁是等待一个事件(计数器变成0), 栅栏在等待其他线程

(3) 应用场景:

1° 并行计算时, 所有子线程都计算完毕后才能进入下一个步骤

(4) java.util.concurrent.CyclicBarrier是一种栅栏, 它可以使一定数量的参与方反复在栅栏位置汇集。 当线程到达栅栏位置时, 调用CyclicBarrier对象的await()方法, 这个方法将会阻塞直到所有线程到达栅栏位置; 如果所有线程都到达了栅栏位置, 栅栏将会打开, 所以线程被释放, 而CyclicBarrier对象将会被重置以便下次使用

(5) 示例

这个示例首先获得了处理器的个数(通过int count = Runtime.getRuntime().availableProcessors();), 于是创建线程的个数也是处理器的个数, 并将这个参数传给CyclicBarrier; 每个线程执行完毕后, 将会在barrier.await();处阻塞直到所有线程完成工作。由于CyclicBarrier构造函数中传进去了一个Runnable对象, 因此所有线程执行完毕后会调用mainBoard.commitNewValues();方法, 然后再调用mainBoard.waitForConvergence();执行结束

(6) 在不涉及IO操作或共享数据访问的计算问题中, 当线程数量为N或N+1时将获得最优的吞吐量, 更多的线程不会带来任何帮助

FutureTask

(1) FutureTask表示的计算是通过Callable实现的, 并且可以处于以下3种状态: 等待运行、正在运行、运行完成

其中, 运行完成既可以是正常结束, 也可以是抛出异常, 还可以是被取消

(2) 如果任务已经完成, 那么get()方法会立即返回结果; 否则将阻塞直到任务进入完成状态

(3) Callable表示的任务可以抛出受检查异常或未受检查异常, 并且任何代码都可能抛出Error;

无论任务代码抛出了什么异常, 都会被封装在ExecutionException中, 在future.get()中重新被抛出