Notes
  • Introduce
  • Go
    • Grammar
      • Basic
      • Goroutines & Channels
      • Test
    • System Library
      • Module
      • sync
      • context
      • net
    • Concurrency in Go
    • The Go Memory Model
    • Code Snippet
  • Rust
    • The Rust Programming Language
    • Rust by Example
  • JAVA
    • Preface
    • Grammar
      • Basic
      • Data Types
      • Operator
      • Exceptions
    • Class Libraries
      • Collection
      • Stream
      • IO
      • NIO
      • RMI
    • Concurrency
      • Preface
      • JMM
      • Synchronized & CAS
      • Deadlock
      • Thread
      • Lock & Condition
      • Utility Class
      • Thread-safe Collection
      • Atomic Class
      • Fork/Join
      • Concurrency Design Patterns
        • Immutable
        • Copy-on-Write
        • ThreadLocal
        • Multitheading If
        • Division
    • JVM
      • Class & Instance Initialization
      • Runtime Data Area
      • Garbage Collection
    • Web Container
      • Tomcat Architecture
      • Jetty Architecture
    • Spring
    • Tuning
      • Programming
  • Computer Science
    • Computer Organization
    • Algorithm
      • Complexity
      • Linear List
      • Sort
      • Binary Search
      • Skip List
      • Hash Table
      • Tree
      • Graph
      • String Matching
      • Bloom Filter
      • Greedy Algorithm
      • Divide and Conquer
      • Back Tracking
      • Dynamic Programming
    • Network Protocol
      • Pysical Layer
      • Data Link Layer
      • Network Layer
      • Transport Layer
      • Application layer
      • HTTP
      • HTTP/2 in Action
    • Operating System
      • Basic
      • System Initialization
      • Diagnostic Tools
      • CPU Diagnosis
      • Memory Diagnosis
      • Disk Diagnosis
      • Network Diagnosis
      • Monitor System
    • Design Patterns
      • UML
      • OOP
      • Principle
      • Refactoring & Specification
      • Creational
        • Singleton
        • Factory
        • Builder
        • Prototype
      • Structural
        • Proxy
        • Bridge
        • Decorator
        • Adapter
        • Facade
        • Composite
        • FlyWeight
      • Behavioral
        • Observer
        • Template Method
        • Strategy
        • State
        • Iterator
        • Chain of Responsibility
    • Distributed System
      • Protocol & Algorithm
      • Transcation
      • Theory
      • Resource Management
      • Scheduling
      • Computing
      • Message Queue
      • Cache
      • Consistent Hashing
  • database
    • InfluxDB
      • In-Memory Index
      • Meta
    • MySQL
      • SQL
      • Architecture
      • Log
      • Transaction
      • Indexing
      • Lock
      • Storage
    • Redis
    • Elasticsearch
      • Local Debug
    • HBase
    • Kafka
    • ZooKeeper
  • Reading
    • RocketMQ
    • 演说之禅
    • So Good They Can't Ignore You
    • 学会提问
    • Lecture
  • Other
    • v2ray
    • Kubernetes
    • Git
    • Maven
    • Anaconda And Conda
    • Fuck! Shit!
      • Remove Final by Reflection
      • Ingress Host
      • ExecuterService submit
  • Open source contribution
Powered by GitBook
On this page
  • Semaphore
  • CountDownLatch
  • CyclicBarrier
  • ThreadPoolExecutor
  • 构造函数
  • 核心逻辑
  • submit
  • Executors
  • Future
  • FutureTask
  • CompletableFuture
  • 概念
  • 创建 CompletableFuture
  • CompletionStage
  • 串行关系
  • AND 汇聚关系
  • OR 汇聚关系
  • 异常处理
  • CompletionService

Was this helpful?

  1. JAVA
  2. Concurrency

Utility Class

PreviousLock & ConditionNextThread-safe Collection

Last updated 5 years ago

Was this helpful?

Semaphore

java.util.concurrent.Semaphore是对的实现。

用于控制同时访问的线程个数。

底层基于 AQS,state 变量存储的是可用的剩余资源。

/**
* 5台机器,8个工人,一台机器只能被一个工人使用。
*/
public class Test {
    public static void main(String[] args) {
        int N = 8;            //工人数
        Semaphore semaphore = new Semaphore(5); //机器数目
        for(int i=0;i<N;i++)
            new Worker(i,semaphore).start();
    }
 
    static class Worker extends Thread{
        private int num;
        private Semaphore semaphore;
        public Worker(int num,Semaphore semaphore){
            this.num = num;
            this.semaphore = semaphore;
        }
 
        @Override
        public void run() {
            try {
                semaphore.acquire();
                System.out.println("工人"+this.num+"占用一个机器在生产...");
                Thread.sleep(2000);
                System.out.println("工人"+this.num+"释放出机器");
                semaphore.release();           
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

CountDownLatch

主要用来解决一个线程等待多个线程的场景。

计数器不能循环利用。当计数器减到0时,再调用 await 会直接通过。

底层用 AQS 实现,state 变量即为设置的 N。

CyclicBarrier

CyclicBarrier 是一组线程之间互相等待。

计数器可以循环利用,而且能够自动重置,一旦计数器减到0,会自动重置到你设置的初始值。

ThreadPoolExecutor

Java 创建线程需要调用操作系统内核 API,然后操作系统分配一系列资源,所以创建线程是一个重量级操作,应该避免频繁创建于销毁。所以需要线程池来解决上述问题。

构造函数

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
  • corePoolSize:最小线程数。

  • maximumPoolSize:最大线程数。

  • keepAliveTiTime & unit:若一个线程空闲了一段时间,并且线程数量大于 corePoolSize,那么空闲的线程就能被回收。

  • workQueue:阻塞的工作队列。

  • threadFactory:自定义如何创建线程,可以指定有意义的线程名字

  • handler:工作队列满了后,提交任务时的拒绝策略。

    • CallerRunsPolicy:提交任务的线程自己去执行。

    • AbortPolicy:默认,直接拒绝,抛出 RejectedExecutionException。

    • DiscardPolicy:直接丢弃。

    • DiscardOldestPolicy:丢弃最老的任务。

默认拒绝策略要慎重使用。如果线程池处理的任务非常重要,建议自定义自己的拒绝策略;并且在实际工作中,自定义的拒绝策略往往和降级策略配合使用。

核心逻辑

ThreadPoolExecutor 有个内部类 Worker,轮询去任务队列拿任务执行。

public class ThreadPoolExecutor extends AbstractExecutorService {
    private final class Worker 
        extends AbstractQueuedSynchronizer implements Runnable { 
               
        public void run() {
            runWorker(this);
        }
    }
    
    final void runWorker(Worker w) {
        try {
            while (task = getTask() != null) {
                try {
                    task.run();
                } finally {
                    task = null;
                }
            }
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }
}

submit

ThreadPoolExecutor 的 execute() 方法不能取得任务结果,但是 submit() 方法可以。

// Runnable 的 run 方法没有返回值,所以这个方法仅能用来判断任务是否结束
Future<?> submit(Runnable task)

// 假设这个方法返回 f,f.get() 的返回值就是传给 submit() 方法的参数 result。
<T> Future<T> submit(Runnable task, T result)

<T> Future<T> submit(Callable<T> task)

当使用 submit 方法时,若提交的任务抛出了异常,只能通过返回的 Future 获取异常。所以如果即使用 submit 方法,又没有保留返回的 Future,则必须保证提交的任务没有抛出异常,不然异常会被吃掉,不会打印出异常栈。

原因:使用 submit 后,会把任务包装成 FutureTask,在执行时会捕获异常,保存在 Future 里面,自然就不会再抛出异常了;而使用 execute 时,任务抛出的异常会直接再次抛出。

所以:若不需要返回结果,建议使用 execute 方法。

Executors

Java 并发包提供了 Executors 可以快速创建线程池。

目前大厂的编码规范中基本上都不建议使用 Executors 了。因为 Executors 提供的很多方法默认使用的都是无界的 LinkedBlockingQueue。

Future

Future 接口有如下方法:

public interface Future<V> {
    // 取消任务
    boolean cancel(boolean mayInterruptIfRunning);
    // 任务是否已取消
    boolean isCancelled();
    // 任务是否已结束
    boolean isDone();
    // 任务执行结果
    V get() throws InterruptedException, ExecutionException;
    // 任务执行结果,支持超时
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

FutureTask

FutureTask实现了Runnable和Future 接口。所以 FutureTask 可以很容易获取线程的执行结果。

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

public class FutureTask<V> implements RunnableFuture<V> {
    public FutureTask(Callable<V> callable)
    public FutureTask(Runnable runnable, V result)
}

以华罗庚的烧水泡茶为例,可以有很多方法解决,比如 Thread.join()、CountDownLatch 等。用 FutureTask 来解决的方法如下:

// 创建任务T2的FutureTask 
FutureTask<String> ft2 = new FutureTask<>(new T2Task()); 

// 创建任务T1的FutureTask 
FutureTask<String> ft1 = new FutureTask<>(new T1Task(ft2)); 

// 线程T1执⾏任务ft1 
Thread T1 = new Thread(ft1); 
T1.start(); 

// 线程T2执⾏任务ft2 
Thread T2 = new Thread(ft2); 
T2.start(); 

// 等待线程T1执⾏结果 
System.out.println(ft1.get());

// T1Task需要执⾏的任务: 洗⽔壶、烧开⽔、泡茶 
class T1Task implements Callable<String> { 
    FutureTask<String> ft2; 
    
    // T1任务需要T2任务的FutureTask 
    T1Task(FutureTask<String> ft2) { 
        this.ft2 = ft2; 
    } 
    
    @Override 
    String call() throws Exception {
        System.out.println("T1:洗⽔壶..."); 
        TimeUnit.SECONDS.sleep(1);
        
        System.out.println("T1:烧开⽔..."); 
        TimeUnit.SECONDS.sleep(15); 
        
        // 获取T2线程的茶叶 
        String tf = ft2.get(); 
        System.out.println("T1:拿到茶叶:"+tf);
        
        System.out.println("T1:泡茶..."); 
        return "上茶:" + tf;
    }
}

// T2Task需要执⾏的任务: 洗茶壶、洗茶杯、拿茶叶 
class T2Task implements Callable<String> {
    @Override
    String call() throws Exception {
        System.out.println("T2:洗茶壶...");
        TimeUnit.SECONDS.sleep(1);
        
        System.out.println("T2:洗茶杯..."); 
        TimeUnit.SECONDS.sleep(2);
        
        System.out.println("T2:拿茶叶..."); 
        TimeUnit.SECONDS.sleep(1); 
        
        return "⻰井";
    }
} 

CompletableFuture

概念

异步化:是利用多线程优化性能这个核心方案得以实施的基础。

Java 8 提供了 CompletableFuture来支持异步编程,Java 9 提供了更加完备的 Flow API,ReactiveX 的 Java 实现是 RxJava,使得在 Java 6就能使用异步编程。

重新实现一遍上一节的烧水泡茶:

//任务1:洗⽔壶->烧开⽔ 
CompletableFuture<Void> f1 = CompletableFuture.runAsync(()->{ 
	System.out.println("T1:洗⽔壶..."); 
	sleep(1, TimeUnit.SECONDS);

	System.out.println("T1:烧开⽔..."); 
	sleep(15, TimeUnit.SECONDS); 
}); 

//任务2:洗茶壶->洗茶杯->拿茶叶 
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(()->{ 
	System.out.println("T2:洗茶壶..."); 
	sleep(1, TimeUnit.SECONDS);

	System.out.println("T2:洗茶杯..."); 
	sleep(2, TimeUnit.SECONDS);

	System.out.println("T2:拿茶叶..."); 
	sleep(1, TimeUnit.SECONDS); return "⻰井"; 
}); 

//任务3:任务1和任务2完成后执⾏:泡茶 
CompletableFuture<String> f3 = f1.thenCombine(f2, (__, tf)->{ 
	System.out.println("T1:拿到茶叶:" + tf); 
	System.out.println("T1:泡茶...");

	return "上茶:" + tf; 
}); 

创建 CompletableFuture

主要使用下面四种静态方法。

  • 可以指定线程池,若不指定,则使用公用的ForkJoinPool线程池。

  • ForkJoinPool 默认线程数为 CPU 核数,可以通过参数-Djava.util.concurrent.ForkJoinPool.common.parallelism设置。

  • 若是 IO 较重的异步操作,建议使用自定义线程池。

  • runAsync 使用 Runnable 接口,没有返回值。

  • supplyAsync 使用 Supplier 接口,有返回值。

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
    public static CompletableFuture<Void> runAsync(Runnable runnable);
    public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor);
}

CompletionStage

CompletableFuture 实现了 CompletionStage 接口。CompletionStage 定义了工作流的关系,工作流的关系主要有:串行关系、并行关系、汇聚关系。汇聚关系又分为:

  • AND 汇聚:所有任务依赖都完成之后才开始执行当前任务。

  • OR 汇聚:依赖的任务只要有一个完成就可以执行当前任务。

CompletionStage 除了定义工作流的关系之外,还要能够处理异常。

串行关系

  • thenApply(Function):既能接受参数,也支持返回值。

  • thenAccept(Consumer):能接受参数,不支持返回值。

  • thenRun(Runnable):不能接受参数,也不支持返回值。

public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor);

public CompletionStage<Void> thenAccept(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor);

public CompletionStage<Void> thenRun(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action, Executor executor);

AND 汇聚关系

三者的差别与上文 Function、Consumer、Runnable 的差别一致。

public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn);
public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn);
public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor);

public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action);
public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action);
public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor);

public CompletionStage<Void> runAfterBoth(CompletionStage<?> other, Runnable action);
public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action);
public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor);

OR 汇聚关系

三者的差别与上文 Function、Consumer、Runnable 的差别一致。

public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn);
public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn);
public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor);
              
public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action);
public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action);
public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor);

public CompletionStage<Void> runAfterEither(CompletionStage<?> other, Runnable action);
public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action);
public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor);

异常处理

  • exceptionally:相当于 catch{}代码块。

  • whenComplete:相当于 finally{}代码块,无论异常是否发生都会执行,不支持返回结果。

  • handleAsync:相当于finally{}代码块,支持返回结果,

public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn);

public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action);
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action);
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor);

public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor);

CompletionService

实现原理:内部维护一个阻塞队列,任务执行结束就把执行结果加入到阻塞队列。

实现类是 ExecutorCompletionService:

public class ExecutorCompletionService<V> implements CompletionService<V> {
    // 默认使用无界队列 LinkedBlockingQueue
    public ExecutorCompletionService(Executor executor)
    
    public ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)
}

CompletionService 有 5 个方法:

public interface CompletionService<V> {
    Future<V> submit(Callable<V> task);

    // 与 ThreadPoolExecutor.submit(Runnable task, T result) 类似
    Future<V> submit(Runnable task, V result);

    // 若阻塞队列为空,则会阻塞
    Future<V> take() throws InterruptedException;

    // 若阻塞队列为空,则返回 null
    Future<V> poll();

    // 超时版的 poll 方法
    Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}
信号量模型