java并发编程day08

以你之姓@ 2022-06-08 07:46 189阅读 0赞

## 5.4 阻塞和可中断的方法 ##

线程可能会因为几种原因被阻塞或暂停：  
\- 等待I/O操作结束  
\- 等待获得一个锁  
\- 等待从Thread.sleep中唤醒  
\- 等待另一个线程的计算结果

当一个线程阻塞时，他通常被挂起，并被设置为线程阻塞的某个状态。  
一个阻塞的操作和一个普通的操作之间的差别仅仅在于：被阻塞的线程必须要等待一个事件的发生才能继续进行，并且这个事件是超越它自己控制的，因为需要花费更长的时间：等待I/O操作完成，锁可用，或者是外部计算结束。当外部事件发生后，线程被置回RUNNABLE状态，重新获得调度的机会。

BlockingQueue的put和take方法会抛出一个受检查的InterruptedException，这与类库中的其他一些方法是相同的，比如Thread.sleep。当一个方法能够抛出InterruptedException的时候，实在告诉你这个方法是一个可阻塞的方法，进一步看，如果它被终端，将可以提前结束阻塞状态。

Thread提供了interrupt方法，用来中断一个线程，或者查询某线程是否已经被中断。没一个线程都有一个boolean的属性，这个属性代表了线程的中断状态；中断线程时需要设置这个值。

中断是一种协作机制。一个线程不能迫使其他线程停止正在做的事情，或者去做其他事情：当线程A中断B时，A仅仅是要求B在达成某个方便停止的关键点时，停止正在做的事情，如果它这样做是正确的。然而，在API或者语法规范中，任何特定应用级别的中断语义中，最常用的是取消一个互动。从时间角度来看，相应中断的阻塞方法，可以更容易的取消耗时的活动。

当你在代码中调用了一个会抛出InterruptedException的方法时，你自己的方法也就成为了一个阻塞方法，要为响应中断做好准备。在类库代码中，有两种基本选择：  
\- 传递InterruptedException。即抛出  
\- 恢复中断。 捕获异常，并且在当前线程中通过调用interrup从中断中恢复，这样调用栈中更高层的代码可以发现中断已经发生。如下代码所示

public class TaskRunnable implements Runnable{
        
        BlockingQueue<Task> queue;
        ...
        public void run(){
            try{
                processTask(queue.take());  
            }catch(InterruptedException e){
                //恢复中断状态
                Thread.currentThread().interrupt();
            }
        }
    }

*  我们不应该捕获异常后什么都不响应。这样做会丢失线程中断的数据。从而剥夺了上层栈的代码处理中断的机会。只有一种情况允许掩盖中断：**扩展Thread** 并因此控制了所有处于调用栈上层的代码、

## 5.5 Synchronizer ##

阻塞队列在容器类中是独一无二的：它们不仅作为对象的容器，而且能够协调生产者线程和消费者线程之间的控制流，这是因为take和put方法会保持阻止状态知道队列进入了期望的状态（不满也不空）

**Synchronizer** 是一个对象，它根据自身的状态调节线程的控制流。阻塞队列可以扮演一个Synchronizer的角色；其他类型的Synchronizer包括信号量、关卡以及闭锁。在平台类库中存在一些synchronizer类；如果这些不能满足需要，我们可以创建一个自己的Synchronizer。

所有Synchronizer都享有类似的结构特性：封装状态，而这些状态决定着线程执行到某一点时是通过还是被迫等待；他们还提供操控状态的方法，以及高效的等待Synchronizer进入到期望状态的方法。

### 5.5.1 闭锁 ###

闭锁是一种Sychronizer，他可以延迟线程的进度知道线程直到终止状态。一个闭锁工作起来就像是一道大门：直到闭锁达到终点状态之前，门一直是关闭的，没有线程能够通过，在终点状态到来的时候，门开了，允许所有线程通过。一旦闭锁到达了终点状态它就不能再改变状态了，所以它会永远保持敞开状态。闭锁可以用来确保特定活动指导其他的活动完成后才发生，比如：

*  确保一个计算不会执行，直到它需要的资源被初始化。
 *  确保一个服务不会开始，直到它依赖的的其他服务都已经开始
 *  等待，直到活动的所有部分都为继续处理做好充分准备，比如王者荣耀的十名玩家都准备就绪然后才开始游戏。

CountDownLatch是一个灵活的闭锁实现，用于上述各种情况：允许一个或多个线程等待一个事件集的发生。闭锁的状态包括一个计数器，初始化为一个证书，用来表现需要等待的事件数。countDown方法对计数器做加减操作，表示一个事件已经发生了，而await方法等待计数器达到0，此时所有需要等待的事件都已发生。如果计数器入口时值为非0，awit会一直阻塞直到计数器为0，或者等待线程中断及超时。

如下面代码，TestHarness阐释了闭锁的两种常见用法。TestHarness创建了一些线程，并发地执行给定的任务。它使用两个闭锁，一个开始阀门，一个结束阀门。 这个开始阀门将计数器初始化为1 ， 结束阀门将计数器初始化为工作线程的数量。每一个工作线程要做的第一件事就是等待开始阀门打开，这样就能确保直到所有线程都做好准备时，才能开始工作。 每个线程的最后一个工作是为结束阀门减一，这样做使控制线程有效的等待，直到最后一个工作线程完成任务，这样就能计算整个的用时。  
在TestHarness中，我们为什么不在线程创建后就立即启动，却要自寻烦恼的使用TestHarness呢？或许是因为我们想要在计算在线程N倍并发地情况下执行一个任务的时间。如果我们简单的创建线程病启动线程，那么先启动的就比后启动的具有领先优势，并且根据活动线程数的增加或者减少，这样的竞争度也在不断改变。开始阀门让控制线程能够在同时释放所有工作者线程，结束阀门让控制线程能够等待最后一个线程完成任务，而不是顺序等待每一个线程结束。

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 e) {
                            e.printStackTrace();
                        }
    
                    };
                };
                t.start();
            }
            long start = System.nanoTime();
            startGate.countDown();
            endGate.await();
            long end = System.nanoTime();
            return end-start;
        }
    }

### 5.5.2 FutureTask ###

FutureTask同样可以作为闭锁。Future的计算是通过Callable实现的，它等价于一个可携带结果的Runnable，并且有三个状态：等待、运行和完成。完成包括所有计算以任务的方式结束，包括正常结束、取消和异常。一旦FutureTask进入完成状态，它会永远停止在这个状态上。

Future.get的行为依赖于任务的状态。如果它已经完成，get可以立刻得到返回结果，否则会被阻塞直到任务转入完成状态，然后会返回结果或者抛出异常。FutureTask把计算的结果从运行计算额线程传送到需要这个结果的线程；FutureTask的规约保证了这种传递建立在结果的安全发布基础之上。  
Executor框架利用FutureTask来完成异步任务，并可以用来进行任何潜在的耗时计算，而且可以在真正需要计算结果之前就启动它们开始计算。如下面代码所示：Preloader使用了FutureTask来执行一个价格昂贵的计算，结果稍后会被用到，尽早开始计算，就可以减少等待结果所需要花费的时间。

public class Preloader {
        
        private final FutureTask<ProductInfo> future = 
                new FutureTask<ProductInfo>(new Callable<ProductInfo>() {
                    public ProductInfo call() throws DataLoadException{
                        return loadProductInfo();
                    }
                });
        private final Thread thread = new Thread(future);
        public void start(){
            thread.start();
        }
        public ProductInfo get() throws DataLoadException,InterruptedException{
            try {
                return future.get();
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                if (cause instanceof DataLoadException) {
                    throw cause;
                }else{
                    throw launderThrowable(cause);
                }
            }
        }
        public static RuntimeException launderThrowable(Throwable t){
                if (t instanceof RuntimeException) {
                    return (RuntimeException) t;
                }else if (t instanceof Error) {
                    throw (Error)t;
                }else{
                    throw new IllegalStateException("not uncheched ",t);
                }
            }
        }

Preloader创建了一个FutureTask，记录从数据库加载结果的信息，还创建了一个将要执行运算的线程。在构造函数或者静态初始化方法中启动线程并不是明智的，所以它提供了start方法启动线程。当程序经过一段时间后需要ProductInfo时，它可以调用get，如果已经加载就绪的话，就会返回这些数据，否则会等待加载结束再返回。  
Callable记录的这些任务，可以抛出受检查或者未受检查的异常，并且任何代码都可能抛出Error，无论执行任务的代码抛出什么，它都被封装为一个ExecutionException，并被Future.get重新抛出，调用get的代码如此复杂，不仅因为必须处理可能出现的ExecutionException和未受检查的CancllationException，而且还因为ExecutionException的诱因是作为Throwable返回的，Throwable处理起来不方便。

在Preloader中，如果get抛出一个ExecutionException，那么原因有以下三种：Callable抛出的受检查的异常，或是一个RuntimeException或者一个Error。我们必须对这三种情况进行分别处理。在调用launderThrowable之前，Preloader先检查已知异常，并重新抛出。这样只留下那些未受检查的异常，Preloader通过调用launderThrowable传递了一个Error，launderThrowable方法会直接再将其抛出。如果不是一个RuntimeException它会抛出一个IllegalStateException用来指定这是一个逻辑错误。这样就只剩下  
RuntimeException了，launderThrowable会把它返回给它的调用者，通常由调用者再将其抛出。

### 5.5.3 信号量 ###

计算信号量用来控制能够同时访问某特定资源的活动的数量，或者同时执行某一给定操作的数量。计数信号量可以用来实现资源池或者给一个容器限定边界。  
一个Semaphore管理一个有效的许可集；许可的初始量通过构造函数传递给Smahore。活动能够获得许可（只要还有剩余许可），并在使用之后释放许可。如果已经没有可以用的许可了，那么acquire会被阻塞，直到有可用的为止（或者直到被中断或者操作超时）。release方法向信号量返回一个许可。计算信号量的一种退化形式是二元信号量；一个计数初始值为1的Semaphore。二元信号量可用作互斥锁，它有不可重入锁的语意；谁拥有这个唯一的许可，就拥有了互斥锁。

信号量可以用来实现资源池，比如数据库连接池。哟一个定长的池，当它为空时，你向它请求资源会失败。构建这种池很容易，然而当池为空时，你真正需要做的事阻塞它，然后在它不为空时，再次解除阻塞。如果你以池的大小初始化一个Semaphore，在你从池中获取资源之前，你应该调用acquire方法获取一个许可，调用release把许可放回资源池。acquire会一直阻塞，直到池不再为空。  
相似的，我们也可以使用Semahore把任何容器转化为有界的阻塞容器，就像BoundedHashSet所示一样。信号量被初始化为容器所期望的最大值。add操作在向底层容器中添加条目之前，先获得一个许可。事实上，如果add操作没有能加入任何东西，它会立即释放一个许可。同样，一个成功的remove操作释放一个许可，使更多的元素能够加入其中。底层的Set实现并不知道边界在哪。这是由BoundedHashSet控制的。

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;
        }
    
    }

### 5.5.4 关卡 ###

关卡类似于闭锁，它们都能阻塞一组线程，直到某些事件发生。  
不同点在于：所有线程必须同时到达关卡点，才能继续处理。闭锁等待的是事件；关卡等待的是其他线程。关卡实现的协议，就像一些家庭成员指定商场中的集合地点，到齐了之后再决定接下做什么。

CyclicBarrier允许一个给定数量的成员多次集中在一个关卡点，这在并行迭代算法中非常有用，这个算法会把一个问题拆分成一系列相互独立的子问题。当线程到达关卡点时，调用await，await会被阻塞，直到所有线程都到达关卡点。如果所有线程都到达了关卡点，关卡就被成功的突破，这样所有线程都被释放，关卡会重置以备下一次使用。如果对await的调用超时，或者阻塞的线程被中断，那么关卡就被任务是失败的，所有对await未完成的调用都通过BrokenBarrierExcetion终止。如果成功的通过了关卡，await为每一个线程返回一个唯一的到达索引号，可以用它来选举一个leader，在下一次迭代中承担一些特殊工作。CyclicBarrier也允许你向构造函数传递一个关卡行为；这是一个Runnable，当成功通过关卡时，会执行，但是在阻塞线程被释放之前是不能执行的。

关卡通常被用来模拟这种情况：一个步骤的计算可以并行完成，按时要求必须完成所有与一个步骤相关的工作后才能进入下一步。如下面代码CellularAuomate演示了使用关卡来计算一个细胞的自动化模拟，比如Conway’s的生命游戏。当模拟处于并行状态时，它通常会不假思索的为每一个元素分配独立的线程，浙江带来太多的线程，协调线程所花费的开销会延缓计算。代替的做法是：把问题划分成不同的子部分，让每一个线程解决其中一部分，之后再合并结果。CellularAutomata把问题分成了与可用CPU数量相同的几部分，并把每个部分分配给一个线程。再每一步中，工作线程都为自己部分的细胞计算一个新值。当所有线程到达关卡时，关卡的操作是想数据模型提交一个新的值。在关卡的操作完成后，工作线程被释放进行下一步计算，这包括询问一个isDone方法来决定是否需要下一次迭代。  
Exchanger是关卡的另一种形式，它是一种两步关卡，在关卡点会交换数据。当两房进行的活动不对称时，Exchanger是非常有用的，比如当一个线程向缓冲写入一个数据，这时另一个线程充当消费者使用这个数据；这些线程可以使用Exchanger进行会面，并用完整成缓冲与空缓冲进行交换。当两个线程通过Exchanger交换对象时，交换我双方的对象建立了一个安全的发布。

交换的时机取决于应用程序的响应需求。最简答的方案是当写入任务的缓冲写满时就发生交换，并且当清除任务的缓冲清空后也发生交换；这样做使交换的次数最少，但是如果新数据的到达率不可测的话，处理一些数据会发生延迟。另一个方案是，缓冲满了就发生交换但是当缓冲部分充满却已经存在了特定长的时间时，也会发生交换。

public class CellularAuomate {
        
        private final Board mainBoard;
        private final CyclicBarrier barrier;
        private final Worker[] workers;
    
        public CellularAuomate(Board board){
            this.mainBoard = board;
            int count = Runtime.getRuntime().availableProcessors();
            this.barrier= new CyclicBarrier(count,new Runnable() {
    
                @Override
                public void run() {
                    mainBoard.commitNewValues();
                }
            });
            this.workers = new Worker[count];
            for (int i = 0; i < count; i++) {
                workers[i] = new Worker(mainBoard.getSubBoard(count,i));
            }
    
    
        }
    
    
        private class Worker implements Runnable{
        
            private final Board board ;
            public Worker(Board board){
                this.board = board;
            }
            @Override
            public void run() {
                while (! board.hasConverged()) {
                    for (int x = 0; x < board.getMaxX(); x++) {
                        for (int y = 0; y < board.getMaxY(); y++) {
                            board.setNewValue(x,y,computValue(x,y));
                            try{
                                barrier.await();
                            }catch (InterruptedException ex){
                                return ;
                            }catch (BrokenBarrierException ex ){
                                return ;
                            }
                        }
                    }
                }
            }
    
        }
        public void start(){
            for (int i = 0; i < workers.length; i++) {
                new Thread(workers[i]).start();
            }
            mainBoard.waitForConvergence();
        }
    }