java.util.concurrent 包介绍

迷南。 2022-09-11 02:25 130阅读 0赞

### 文章目录 ###

*  1.介绍
 *  2.主要组件
 *   *  2.1. Executor
     *  2.2. ExecutorService
     *  2.3. ScheduledExecutorService
     *  2.4. Future
     *  2.5. CountDownLatch
     *  2.6. CyclicBarrier
     *  2.7. Semaphore
     *  2.8. ThreadFactory
     *  2.9.BlockingQueue
     *  2.10. DelayQueue
     *  2.11. Locks
     *  2.12. Phaser

# 1.介绍 #

java.util.concurrent 包提供了用于创建并发应用程序的工具。

# 2.主要组件 #

java.util.concurrent 包含太多的特性。 在本篇中，主要关注此包中一些最有用的实用程序，

例如：

*  *Executor*
 *  *ExecutorService*
 *  *ScheduledExecutorService*
 *  *Future*
 *  *CountDownLatch*
 *  *CyclicBarrier*
 *  *Semaphore*
 *  *ThreadFactory*
 *  *BlockingQueue*
 *  *DelayQueue*
 *  *Locks*
 *  *Phaser*

## 2.1. Executor ##

Executor是一个表示提供任务对象的接口。

如果任务要在新线程或当前线程上运行，这取决于特定的实现（从哪里启动调用）。 因此，使用此接口，可以将任务执行流程与实际任务执行机制解耦。

这里需要注意的一点是，Executor 并没有严格要求任务执行是异步的。 在最简单的情况下，执行者可以在调用线程中立即调用提交的任务。

创建一个调用程序实例：

public class Invoker implements Executor { 
        @Override
        public void execute(Runnable r) { 
            r.run();
        }
    }

调用程序来执行任务。

public void execute() { 
        Executor executor = new Invoker();
        executor.execute(() -> System.out.println("execute task ..."));
    }

这里需要注意的是，如果executor不能接受任务执行，就会抛出RejectedExecutionException。

## 2.2. ExecutorService ##

ExecutorService 是一个完整的异步处理解决方案。 它管理内存队列并根据线程可用性安排提交的任务。

要使用 ExecutorService，需要创建一个 Runnable 类。

public class Task implements Runnable { 
        @Override
        public void run() { 
            // 任务细节
        }
    }

现在创建 ExecutorService 实例并分配这个任务。 在创建时，需要指定线程池大小。

ExecutorService executor = Executors.newFixedThreadPool(10);

如果想创建一个单线程的ExecutorService实例，可以使用newSingleThreadExecutor(ThreadFactory threadFactory)来创建该实例。

一旦创建了执行器，就可以使用它来提交任务。

public void execute() {  
        executor.submit(new Task()); 
    }

还可以在提交任务时创建 Runnable 实例。

executor.submit(() -> { 
        new Task();
    });

它还带有两种开箱即用的执行终止方法。 第一个是shutdown(); 它等待所有提交的任务完成执行。 另一种方法是shutdownNow()，它立即终止所有挂起/正在执行的任务。

还有一个方法 awaitTermination(long timeout, TimeUnit unit) 强制阻塞直到所有任务在触发关闭事件或执行超时后完成执行，或者执行线程本身被中断，

try { 
        executor.awaitTermination( 20l, TimeUnit.NANOSECONDS );
    } catch (InterruptedException e) { 
        e.printStackTrace();
    }

## 2.3. ScheduledExecutorService ##

ScheduledExecutorService 是一个类似于 ExecutorService 的接口，但它可以定期执行任务。

Executor 和 ExecutorService 的方法是现场调度的，不会引入任何人为的延迟。 零或任何负值表示需要立即执行请求。

可以同时使用 Runnable 和 Callable 接口来定义任务。

public static void execute() throws Exception { 
            ScheduledExecutorService executorService
                    = Executors.newSingleThreadScheduledExecutor();
    
            Future<String> future = executorService.schedule(() -> "Hello world", 20, TimeUnit.SECONDS);
    
            ScheduledFuture<?> scheduledFuture = executorService.schedule(() -> { 
                System.out.println("延迟5s执行");
            }, 5, TimeUnit.SECONDS);
    
            while (future.isDone() && !future.isCancelled()) { 
                Thread.sleep(2000);
            }
            System.out.println(future.get());
    
            executorService.shutdown();
        }

ScheduledExecutorService 还可以在给定的固定延迟后执行任务：

executorService.scheduleAtFixedRate(() -> { 
        // ...
    }, 1, 10, TimeUnit.SECONDS);
    
    executorService.scheduleWithFixedDelay(() -> { 
        // ...
    }, 1, 10, TimeUnit.SECONDS);

scheduleAtFixedRate( Runnable command, long initialDelay, long period, TimeUnit unit ) 方法创建并执行一个周期性动作，该动作首先在提供的初始延迟后调用，然后在给定的时间段内调用，直到服务实例关闭。

scheduleWithFixedDelay( Runnable command, long initialDelay, long delay, TimeUnit unit ) 方法创建并执行一个周期性动作，该动作在提供的初始延迟后首先被调用，并在执行的终止和调用之间重复给定的延迟 下一个。

## 2.4. Future ##

Future 用于表示异步操作的结果。 它带有用于检查异步操作是否完成、获取计算结果等的方法。

更重要的是，cancel(boolean mayInterruptIfRunning) API 取消操作并释放正在执行的线程。 如果 mayInterruptIfRunning 的值为 true，则执行任务的线程将立即终止。

否则，将允许完成正在进行的任务。

public void invoke() { 
        ExecutorService executorService = Executors.newFixedThreadPool(10);
    
        Future<String> future = executorService.submit(() -> { 
            // 执行某些任务
            Thread.sleep(10000l);
            return "Hello world";
        });
    }

可以使用以下代码片段来检查Future的结果是否准备就绪，如果计算完成则获取数据：

if (future.isDone() && !future.isCancelled()) { 
        try { 
            str = future.get();
        } catch (InterruptedException | ExecutionException e) { 
            e.printStackTrace();
        }
    }

还可以为给定操作指定超时。 如果任务花费的时间超过这个时间，则会抛出 TimeoutException：

try { 
        future.get(10, TimeUnit.SECONDS);
    } catch (InterruptedException | ExecutionException | TimeoutException e) { 
        e.printStackTrace();
    }

## 2.5. CountDownLatch ##

CountDownLatch（在 JDK 5 中引入）是一个实用程序类，它阻塞一组线程直到某些操作完成。

一个 CountDownLatch 用一个计数器（整数类型）初始化； 该计数器随着依赖线程完成执行而递减。 但是一旦计数器达到零，其他线程就会被释放。

## 2.6. CyclicBarrier ##

CyclicBarrier 的工作原理与 CountDownLatch 几乎相同，只是可以重用它。 与 CountDownLatch 不同，它允许多个线程在调用最终任务之前使用 await() 方法（称为屏障条件）相互等待。

public class CyclicBarrierTest { 
        public static void main(String[] args) { 
            start();
        }
    
        public static void start() { 
    
            CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> { 
                // ...
                System.out.println("All previous tasks are completed");
            });
    
            Thread t1 = new Thread(new Task(cyclicBarrier), "T1");
            Thread t2 = new Thread(new Task(cyclicBarrier), "T2");
            Thread t3 = new Thread(new Task(cyclicBarrier), "T3");
    
            if (!cyclicBarrier.isBroken()) { 
                t1.start();
                t2.start();
                t3.start();
            }
    
        }
    }
    
    class Task implements Runnable { 
    
        private CyclicBarrier barrier;
    
        public Task(CyclicBarrier barrier) { 
            this.barrier = barrier;
        }
    
        @Override
        public void run() { 
            try { 
                System.out.println(Thread.currentThread().getName() +
                        " is waiting");
                barrier.await();
                System.out.println(Thread.currentThread().getName() +
                        " is released");
            } catch (InterruptedException | BrokenBarrierException e) { 
                e.printStackTrace();
            }
        }
    
    }

输出

T1 is waiting
    T2 is waiting
    T3 is waiting
    All previous tasks are completed
    T3 is released
    T1 is released
    T2 is released

isBroken() 方法检查在执行期间是否有任何线程被中断。 在执行实际过程之前，我们应该始终执行此检查。

## 2.7. Semaphore ##

Semaphore(信号量)用于阻止线程级访问物理或逻辑资源的某些部分。 信号量包含一组许可； 每当一个线程试图进入临界区时，它需要检查信号量是否有许可可用。

如果许可不可用（通过 tryAcquire()），则不允许线程跳入临界区； 但是，如果许可可用，则准予访问，并且许可计数器减少。

一旦执行线程释放临界区，许可计数器再次增加（由 release() 方法完成）。

可以使用 tryAcquire(long timeout, TimeUnit unit) 方法指定获取访问的超时时间。

还可以检查可用许可的数量或等待获取信号量的线程数量。

public class SemaphoreTest { 
        static Semaphore semaphore = new Semaphore(10);
    
        public static void main(String[] args) throws InterruptedException { 
            for (int i = 0; i < 20; i++) { 
                Thread.sleep(500);
                Thread thread = new Thread(() -> { 
                    try { 
                        execute();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
                thread.setName("Thread[" + i + "]");
                thread.start();
    
            }
        }
    
    
        public static void execute() throws InterruptedException { 
    
            System.out.println("可用的信号量 : " + semaphore.availablePermits());
            System.out.println("等待获取的线程数: " +
                    semaphore.getQueueLength());
    
            if (semaphore.tryAcquire()) { 
                try { 
                    System.out.println(Thread.currentThread().getName() + "：正在执行任务...");
                    Thread.sleep(new Random().nextInt(10) * 1000);
                } finally { 
                    semaphore.release();
                }
            } else { 
                System.out.println(Thread.currentThread().getName() + "：没有执行任务权限");
            }
    
        }
    }

可以使用信号量实现类似互斥锁的数据结构。

## 2.8. ThreadFactory ##

ThreadFactory 充当线程（不存在）池，可按需创建新线程。 它消除了实现高效线程创建机制所需的大量样板代码。

public class CustomThreadFactory  implements ThreadFactory { 
        private int threadId;
        private String name;
    
        public CustomThreadFactory(String name) { 
            threadId = 1;
            this.name = name;
        }
    
        @Override
        public Thread newThread(Runnable r) { 
            Thread t = new Thread(r, name + "-Thread_" + threadId);
            System.out.println("新建线程id为 : " + threadId +
                    " name为 : " + t.getName());
            threadId++;
            return t;
        }
    
        public static void main(String[] args) { 
            CustomThreadFactory factory = new CustomThreadFactory(
                    "CustomThreadFactory");
            for (int i = 0; i < 10; i++) { 
                int finalI = i;
                Runnable runnable = () -> System.out.println(finalI + " runable thread");
                Thread t = factory.newThread(runnable);
                t.start();
            }
        }
    }

## 2.9.BlockingQueue ##

在异步编程中，最常见的集成模式之一是生产者-消费者模式。 java.util.concurrent 包带有一个称为 BlockingQueue 的数据结构——它在这些异步场景中非常有用。

## 2.10. DelayQueue ##

DelayQueue 是一个无限大小的元素阻塞队列，其中一个元素只有在它的到期时间（称为用户定义的延迟）完成时才能被拉取。 因此，最顶部的元素（头部）将具有最大的延迟量，并且将最后轮询。

## 2.11. Locks ##

Lock 是一个实用的类，用于阻止其他线程访问特定代码段，除了当前正在执行它的线程。

Lock 和 Synchronized 块之间的主要区别在于，同步块完全包含在方法中； 但是，可以在不同的方法中使用 Lock API 的 lock() 和 unlock() 操作。

## 2.12. Phaser ##

Phaser 是一个比 CyclicBarrier 和 CountDownLatch 更灵活的解决方案——用作可重用的屏障，动态数量的线程在继续执行之前需要等待。 可以协调执行的多个阶段，为每个程序阶段重用一个 Phaser 实例。