Java中使用CompletableFuture处理异步超时

一天,我在改进多线程代码时被Future.get()卡住了。

一. Future

JDK
5引入了Future模式。Future接口是Java多线程Future模式的实现,在java.util.concurrent包中,可以来进行异步计算。

Future模式是多线程设计常用的一种设计模式。Future模式可以理解成:我有一个任务,提交给了Future,Future替我完成这个任务。期间我自己可以去做任何想做的事情。一段时间之后,我就便可以从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;
}

Future接口的方法介绍如下:

  • boolean cancel (boolean mayInterruptIfRunning)
    取消任务的执行。参数指定是否立即中断任务执行,或者等等任务结束
  • boolean isCancelled ()
    任务是否已经取消,任务正常完成前将其取消,则返回 true
  • boolean isDone ()
    任务是否已经完成。需要注意的是如果任务正常终止、异常或取消,都将返回true
  • V get () throws InterruptedException, ExecutionException
    等待任务执行结束,然后获得V类型的结果。InterruptedException
    线程被中断异常,
    ExecutionException任务执行异常,如果任务被取消,还会抛出CancellationException
  • V get (long timeout, TimeUnit unit) throws InterruptedException,
    ExecutionException, TimeoutException
    同上面的get功能一样,多了设置超时时间。参数timeout指定超时时间,uint指定时间的单位,在枚举类TimeUnit中有相关的定义。如果计
    算超时,将抛出TimeoutException

一般情况下,我们会结合Callable和Future一起使用,通过ExecutorService的submit方法执行Callable,并返回Future。

        ExecutorService executor = Executors.newCachedThreadPool();

        Future<String> future = executor.submit(() -> { //Lambda 是一个 callable, 提交后便立即执行,这里返回的是 FutureTask 实例
            System.out.println("running task");
            Thread.sleep(10000);
            return "return task";
        });

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }

        System.out.println("do something else");  //前面的的 Callable 在其他线程中运行着,可以做一些其他的事情

        try {
            System.out.println(future.get());  //等待 future 的执行结果,执行完毕之后打印出来
        } catch (InterruptedException e) {
        } catch (ExecutionException e) {

        } finally {
            executor.shutdown();
        }

比起future.get(),其实更推荐使用get (long timeout, TimeUnit unit)
方法,设置了超时时间可以防止程序无限制的等待future的结果。

Overview

本节主要介绍异常任务最核心的概念:线程。

public void serve() throws InterruptedException, ExecutionException, TimeoutException {
  final Future<Response> responseFuture = asyncCode();
  final Response response = responseFuture.get(1, SECONDS);
  send(response);
}
private void send(Response response) {
  //...
}

二. CompletableFuture介绍

Java

这是用Java写的一个Akka应用程序,使用了一个包含1000个线程的线程池(原来如此!)——所有的线程都在阻塞在这个
get()
中。系统的处理速度跟不上并发请求的数量。重构以后,我们干掉了所有的这些线程仅保留了一个,极大的减少了内存的占用。我们简单一点,通过一个Java
8的例子来演示。第一步是使用CompletableFuture来替换简单的Future(见:Tip
9)。

2.1 Future模式的缺点

  • Future虽然可以实现获取异步执行结果的需求,但是它没有提供通知的机制,我们无法得知Future什么时候完成。

  • 要么使用阻塞,在future.get()的地方等待future返回的结果,这时又变成同步操作。要么使用isDone()轮询地判断Future是否完成,这样会耗费CPU的资源。

Thread

线程主要用于执行并发任务,提高 CPU 的运行效率。在Java
中线程相关的概念主要有两个: Thread 和 Runnable。

Runnable 是一个只含有 run()
方法的接口,该接口定义了线程所要执行的代码。Thread 表示线程,本身实现了
Runnable 接口,可以执行自己的 run() 方法也可以执行传入的 Runnable
接口的 run() 方法。

例:

定义一个简单的线程

new Thread(() -> {
    System.out.println("async");
}, "thread-01").start();
System.out.println("sync");
  • 通过控制任务提交到ExecutorService的方式:只需用
    CompletableFuture.supplyAsync(…, executorService) 来代替
    executorService.submit(…) 即可
  • 处理基于回调函数的API:使用promises

2.2 CompletableFuture介绍

Netty、Guava分别扩展了Java 的 Future 接口,方便异步编程。

Java 8新增的CompletableFuture类正是吸收了所有Google
Guava中ListenableFuture和SettableFuture的特征,还提供了其它强大的功能,让Java拥有了完整的非阻塞编程模型:Future、Promise
和 Callback(在Java8之前,只有无Callback 的Future)。

CompletableFuture能够将回调放到与任务不同的线程中执行,也能将回调作为继续执行的同步函数,在与任务相同的线程中执行。它避免了传统回调最大的问题,那就是能够将控制流分离到不同的事件处理器中。

CompletableFuture弥补了Future模式的缺点。在异步的任务完成后,需要用其结果继续操作时,无需等待。可以直接通过thenAccept、thenApply、thenCompose等方式将前面异步处理的结果交给另外一个异步事件处理线程来处理。

Thread Pool

如果总是建立新的线程,CPU
资源很快会被耗尽,因此便有了线程池。线程池本质是预先建立好一些空闲线程放在池子中,当需要在线程中执行代码时便从池中取出空闲的线程来运行,运行完毕后线程继续放回池中,如果需要执行代码但是池中没有线程时根据不同的线程池的策略或者新建线程或者将请求放在等待队列中。

Java 中创建线程池非常简单,主要使用 Executors 这一个工具类。

例:

创建一个单线程的线程池

ExecutorService service = Executors.newSingleThreadExecutor();
service.execute(() -> System.out.println("Running in a pool."));

否则(如果你已经使用了阻塞式的API或
Future<T>)会导致很多线程被阻塞。这就是为什么现在这么多异步的API都让人很烦了。所以,让我们重写之前的代码来接收CompletableFuture:

三. CompletableFuture特性

Callable 与 Future

Callable 与 Future 主要用于从线程中获得返回结果。Callable
用于在线程中执行计算并返回结果,Future 用于等待线程的结果。

例:

ExecutorService service = Executors.newSingleThreadExecutor();
Callable<String> callable = () -> "Hello World";
Future<String> future = service.submit(callable);
String result = future.get();
System.out.println(result);
service.shutdown();
public void serve() throws InterruptedException, ExecutionException, TimeoutException {
    final CompletableFuture<Response> responseFuture = asyncCode();
    final Response response = responseFuture.get(1, SECONDS);
    send(response);
}

3.1 CompletableFuture的静态工厂方法

方法名 描述
runAsync(Runnable runnable) 使用ForkJoinPool.commonPool()作为它的线程池执行异步代码。
runAsync(Runnable runnable, Executor executor) 使用指定的thread pool执行异步代码。
supplyAsync(Supplier<U> supplier) 使用ForkJoinPool.commonPool()作为它的线程池执行异步代码,异步操作有返回值
supplyAsync(Supplier<U> supplier, Executor executor) 使用指定的thread pool执行异步代码,异步操作有返回值

runAsync 和 supplyAsync
方法的区别是runAsync返回的CompletableFuture是没有返回值的。

        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println("Hello");
        });

        try {
            future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        System.out.println("CompletableFuture");

而supplyAsync返回的CompletableFuture是由返回值的,下面的代码打印了future的返回值。

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        System.out.println("CompletableFuture");

Timer

Timer 用于执行定时任务,也可以用 ScheduledThreadPool 进行替代。

例:

Timer timer = new Timer("timer", true);
TimerTask task = new TimerTask() {
    @Override
    public void run() {
        System.out.println("ping!");
    }
};
timer.schedule(task, 100, 100);
Thread.sleep(1000);
timer.cancel();

很明显,这不能解决任何问题,我们还必须利用新的风格来编程:

3.2 Completable

方法名 描述
complete(T t) 完成异步执行,并返回future的结果
completeExceptionally(Throwable ex) 异步执行不正常的结束

future.get()在等待执行结果时,程序会一直block,如果此时调用complete(T
t)会立即执行。

        CompletableFuture<String> future  = CompletableFuture.supplyAsync(() -> "Hello");

        future.complete("World");

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

执行结果:

World

可以看到future调用complete(T t)会立即执行。但是complete(T
t)只能调用一次,后续的重复调用会失效。

如果future已经执行完毕能够返回结果,此时再调用complete(T t)则会无效。

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        future.complete("World");

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

执行结果:

Hello

如果使用completeExceptionally(Throwable
ex)则抛出一个异常,而不是一个成功的结果。

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

        future.completeExceptionally(new Exception());

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

执行结果:

java.util.concurrent.ExecutionException: java.lang.Exception
...

CompletableFuture

public void serve() {
    final CompletableFuture<Response> responseFuture = asyncCode();
    responseFuture.thenAccept(this::send);
}

概念

CompletableFuture 是 Java 1.8
新加入的类,提供了事件驱动的编程模型,能够大幅简化异步调用的繁琐过程从而避免陷入回调地狱。

CompletableFuture 继承自 Future,所以也表示未来的值,但是
CompletableFuture
表示的未来的值可以是预期的,即你可以明确表明你知道将会发生什么。

这个功能上是等同的,但是 serve()
只会运行一小段时间(不会阻塞或等待)。只需要记住:this::send 将会在完成
responseFuture
的同一个线程内执行。如果你不想花费太大的代价来重载已经存在的线程池或send()方法,可以考虑通过
thenAcceptAsync(this::send, sendPool)
好极了,但是我们失去了两个重要属性:异常传播与超时。异常传播很难实现,因为我们改变了API。当serve()存在的时候,异步操作可能还没有完成。如果你关心异常,可以考虑返回
responseFutureor
或者其他可选的机制。至少,应该有异常的日志,否则该异常就会被吞噬了。

创建 CompletableFuture

创建一个最简单的 CompletableFuture 的实例

CompletableFuture<Integer> firstFuture = new CompletableFuture<>();
firstFuture.complete(10);
int value = firstFuture.get();
System.out.println(value);  //  10

以上调用 complete() 表示我知道这个 Future 会返回
10。这个例子是最简单的同步调用方式,下面使用工厂类创建一个支持异步调用的
CompletableFuture 实例。

 ExecutorService service = Executors.newFixedThreadPool(10);
 CompletableFuture<Integer> secondFuture = CompletableFuture.supplyAsync(() -> {
      //  long running
      return new Random().nextInt(1000);
  }, service);

以上 supplyAsync() 创建了一个支持异步调用的 CompletableFuture,该
CompletableFuture 的任务运行在指定的线程池中。

final CompletableFuture<Response> responseFuture = asyncCode();
responseFuture.exceptionally(throwable -> {
    log.error("Unrecoverable error", throwable);
    return null;
});

在同一个 CompletableFuture 上执行链式操作

CompletableFuture 的用法类似 Javascript 的 Promise 编程的概念。使用
thenApplyAsync()thenApply() 可以在同一个 CompletableFuture
上可以执行多个链式操作,前一个操作完成后才能执行下一个操作。

例:

CompletableFuture<String> thirdFuture = secondFuture.thenApplyAsync(integer -> {
    if (integer < 900) {
        throw new IllegalArgumentException();
    }
    return "" + integer;
});

请小心上面的代码:exceptionally()
试图从失败中恢复过来,返回一个可选的结果。这个地方虽可以正常的工作,但是如果对
exceptionally()和withthenAccept() 使用链式调用,即使失败了也还是会调用
send() 方法,返回一个null参数,或者任何其它从 exceptionally()
方法中返回的值。

完成 CompletableFuture

thenAcceptAsync()thenAccept() 可以在一个 CompletableFuture
完成所有操作后执行收尾操作。

CompletableFuture<Void> lastFuture = thirdFuture.thenAcceptAsync(s -> System.out.println("Result is " + s));
responseFuture
    .exceptionally(throwable -> {
        log.error("Unrecoverable error", throwable);
        return null;
    })
    .thenAccept(this::send);  //probably not what you think

异常处理

exceptionally() 方法会在 CompletableFuture
执行中发生异常时被调用,可以在该方法中处理异常。

CompletableFuture<String> safe1 = thirdFuture.exceptionally(throwable -> {
        //  throwable is CompletionException
        if (throwable != null && (throwable.getCause() instanceof IllegalArgumentException)) {
            return "Too small.";
        } else if (throwable != null) {
            return throwable.getMessage();
        }
        return null;
    });

或者也可以使用 handleAsync()
方法,无论异常是否发生,该方法都会被调用,所以即可以在返回前处理异常也可以处理返回值。

例:

CompletableFuture<String> safe2 = thirdFuture.handleAsync((s, throwable) -> {
    //  throwable is CompletionException
    if (throwable != null && (throwable.getCause() instanceof IllegalArgumentException)) {
        return "Result is too small.";
    } else if (throwable != null) {
        return throwable.getMessage();
    }
    return s;
});

丢失一秒超时的问题非常巧妙。我们原始的代码在Future完成之前最多等待(阻塞)1秒,否则就会抛出
TimeoutException。我们丢失了这个功能,更糟糕的是,单元测试超时的不是很方便,经常会跳过这个环节。为了维持超时机制,而又不破坏事件驱动的原则,我们需要建立一个额外的模块:一个在给定时间后必定会失败的
Future。

Groovy

使用同 Java,但是由于 Groovy 暂不支持 Java 1.8
的新增特性,所以一些代码写起来会比 Java 麻烦。

public static <T> CompletableFuture<T> failAfter(Duration duration) {
    final CompletableFuture<T> promise = new CompletableFuture<>();
    scheduler.schedule(() -> {
        final TimeoutException ex = new TimeoutException("Timeout after " + duration);
        return promise.completeExceptionally(ex);
    }, duration.toMillis(), MILLISECONDS);
    return promise;
}

private static final ScheduledExecutorService scheduler =
        Executors.newScheduledThreadPool(
                1,
                new ThreadFactoryBuilder()
                        .setDaemon(true)
                        .setNameFormat("failAfter-%d")
                        .build());

Scala

这个很简单:我们创建一个promise(没有后台任务或线程池的
Future),然后在给定的 java.time.Duration 之后会抛出 TimeoutException
异常。如果在某个地方调用 get() 获取这个
Future,阻塞的时间到达这个指定的时间后会抛出 TimeoutException。

Thread

用法同 Java。

实际上,它是一个包装了 TimeoutException 的
ExecutionException,这个无需多说。注意,我使用了固定一个线程的线程池。这不仅仅是为了教学的目的:这是“1个线程应当能满足任何人的需求”的场景。failAfter()
本身没多大的用处,但是如果和 ourresponseFuture
一起使用,我们就能解决这个问题了。

ExecutionContext

ExecutionContext 可以看做 Java 中的线程池,实际上其本质也是通过 Java 的
ExecutorService 来实现的。

例:

val executor = Executors.newFixedThreadPool(1)
implicit val service = ExecutionContext.fromExecutor(executor)
service.execute(new Runnable {
  override def run(): Unit = {
    println("Running in a pool.")
  }
})

以上例子定义了一个容量为 1 的线程池,并且通过隐式转换(具体见 Scala
的隐式转换一章)使 ExecutionContext 运行在该线程池中。

除了自定义线程池之外,Scala 还有一个全局的默认线程池
scala.concurrent.ExecutionContext.Implicits.global,通常来说这个线程池仅用于测试。因为你不知道其它人或者你引用的其它库中是否有使用默认的线程池进行大量长时间操作。

final CompletableFuture<Response> responseFuture = asyncCode();
final CompletableFuture<Response> oneSecondTimeout = failAfter(Duration.ofSeconds(1));
responseFuture
        .acceptEither(oneSecondTimeout, this::send)
        .exceptionally(throwable -> {
            log.error("Problem", throwable);
            return null;
        });

Future

Scala 中的 Future 表示尚未完成的计算结果,执行在指定的 ExecutionContext
中。一个 Future 一旦被赋值后就不可被改变,可以用于编写高效的并行代码。

这里还做了很多其他事情。在后台的任务接收 responseFuture
时,我们也创建了一个“合成”的 oneSecondTimeout
future,这在成功的时候永远不会执行,但是在1秒后就会导致任务失败。现在我们联合这两个叫做
acceptEither,这个操作将执行先完成 Future 的代码块,而简单的忽略
responseFuture 或 oneSecondTimeout 中运行比较慢的那个。如果 asyncCode()
代码在1秒内执行完成,this::send 就会被调用,而 oneSecondTimeout
异常就不会抛出。但是,如果 asyncCode() 执行真的很慢,oneSecondTimeout
异常就先抛出。由于一个异常导致任务失败,exceptionallyerror
处理器就会被调用,而不是 this::send 方法。你可以选择执行 send() 或者
exceptionally,但是不能两个都执行。当如,如果我们有两个“普通”的 Future
正常执行完成了,则最先响应的那个将调用 send() 方法,后面的就会被丢弃。

创建 Future

使用 Future{} 创建 Future 对象

val s = "Hello"
val f1: Future[String] = Future {
  s + " future!"
}

这个不是最清晰的解决方案。更清晰的方案是包装原始的
Future,然后保证它能在给定的时间内执行。这种操作对
com.twitter.util.Future 是可行的(Scala叫做 within()),但是
scala.concurrent.Future
中没有这个功能(据推测是为了鼓励使用前面的方式)。我们暂时不讨论Scala背后如何执行的,先实现类似
CompletableFuture 的操作。它接受一个 Future 作为输入,然后返回一个
Future,这个 Future 在后台任务完成时候执行完成。但是,如果底层的 Future
执行的时间太长,就或抛出异常:

回调

Future 支持三种回调方式

  • onSuccess 用于在 Future 完成后对计算结果进行某些操作
  • onFailure 用于处理在 Future 计算过程中发生的异常
  • onComplete 用于在 Future 处理完成后进行某些操作,其接收的参数为
    Try[T] 类型,子类为 Success[T] 和
    Failure[T],所以可以在其方法体内同时完成成功和失败的后续处理。

例:

val s = "Hello"
//  使用 Future{} 创建 Future 对象
val f1: Future[String] = Future {
  s + " future!"
}
f1 onSuccess {
    case msg =>
        println(msg)
}
f1 onFailure {
    case e =>
        e.printStackTrace()
}
f1 onComplete {
    case Success(msg) => println(s"Success $msg")
    case Failure(e) => println(s"Failure ${e.getMessage}")
}

一个 Future 上 可以定义多个 onSuccess
等类型的回调方法,但是所有的回调方法返回值都是 Unit
所以无法进行链式操作,同时执行多个回调的顺序的也不保证。

例:

val f2 = Future {
    Thread.sleep(10)
}
f2 onSuccess {
    case _ => println("a")
}
f2 onSuccess {
    case _ => println("b")
}
f2 onSuccess {
    case _ => println("c")
}
public static <T> CompletableFuture<T> within(CompletableFuture<T> future, Duration duration) {
    final CompletableFuture<T> timeout = failAfter(duration);
    return future.applyToEither(timeout, Function.identity());
}

组合

Callback 虽然方便,但是在复杂场景下光使用 Callback 很容易陷入回调地狱。

最简单的回调地狱

例:

Future {
    1
} onSuccess {
    case x => Future {
        x + 3
    } onSuccess {
        case x => Future {
            x * 10
        } onSuccess {
            case x =>
                println(s"Result of callback hell is $x") //  40
        }
    }
}

为此 Scala 为 Future 提供了多种组合方式。

map

map 用于在 Future 执行完毕后产生新的 Future
并执行后续操作,无论异常是否会发生,Future 的结果都会进行传递。

例:

val f1 = Future {
    1   //  or 1/0
}
val f2 = f1 map { x =>
    x + 3
}
f2 map { x =>
    x * 10
} onComplete {
    case Success(x) => println(s"Result of map is $x") //  40
    case Failure(e) => e.printStackTrace()
}

Recovery

recover 用于在 Future 执行过程中发生异常时进行一些拯救工作。

例:

Future {
    1 / 0
} recover {
    case e: ArithmeticException => 0
} map { x =>
    x + 3
} map { x =>
    x * 10
} onComplete {
    case Success(x) => println(s"Result of recover is $x") //  30
    case Failure(e) => e.printStackTrace()
}

Fallback

fallback 用于在 Future 执行过程中发生异常时执行另一个
Future。如果异常没有发生,原始 Future 的结果会被返回。如果原始 Future
发生异常,新 Future 成功执行则返回新 Future 的结果,如果新 Future
中也发生异常,则原始 Future 的异常会被返回。

例:

val f3 = Future {
    1 / 0
}
val f4 = Future {
    10
}
f3 fallbackTo f4 map { x =>
    x + 3
} onComplete {
    case Success(x) => println(s"Result of fallback is $x") //  13
    case Failure(e) => e.printStackTrace()
}

Then

then 听起来容易让人混淆,以为是连续多个
Future,前一个执行完的结果会被传递到后一个。但是实际 then
是为了解决副作用的问题,即 then 连接的多个 Furtue 总是返回初始 Future
的结果。

例:

Future {
    1
} andThen {
    case Success(x) =>
        x + 1
} andThen {
    case Success(x) =>
        x + 10
} andThen {
    case Success(x) =>
        println(s"Result of then is $x") //  1
}

由以上例子可知 x 的值并没有发生改变。

这引导我们实现最终的、清晰的、灵活的方法:

Blocking

Future 可以像 Java 的 Future 一样通过阻塞操作获得执行结果。

例:

val f1 = Future {
  1
}
val result = Await result(f1, Duration(3, TimeUnit.SECONDS))
println(result)
final CompletableFuture<Response> responseFuture = within(
        asyncCode(), Duration.ofSeconds(1));
responseFuture
        .thenAccept(this::send)
        .exceptionally(throwable -> {
            log.error("Unrecoverable error", throwable);
            return null;
        });

Promise

Promise 与 Future 不同,Future 只提供了读取计算值的接口,而 Promise
是一个拥有单个 Future 的容器,提供了写入计算值的接口。Java 1.8 提供的
CompletableFuture 就与 Promise 有些相似。

Promise 可以通过调用 success()complete()failure()
来直接调用其内部 Future 对应的 onSuccess() 等方法。

例:

val p1 = Promise[Int]()
val f1 = p1.future
Future {
  val i = Random.nextInt(100)
  p1.success(i)
}
Future {
  f1 onSuccess {
    case x => println(x)
  }
}

Promise 在计算完成后就不允许再进行计算,否则会抛出异常,所以可以使用
tryXXX() 版本来替换以上版本来避免异常的发生。

p1.trySuccess(1)

Promise 也可以使用 completeWith() 方法来调用其它 Future 来代替其内置的
Future 进行计算

例:

 def heavyFuture = {
  val p = Promise[Int]()
  val f = Future {
    1000
  }
  p completeWith f
  p.future
}

val f = heavyFuture
f onSuccess { case x => println(x) } // 1000

希望你喜欢这篇文章,因为你已经知道在Java里,实现响应式编程不再是什么问题。

SyncVar

SyncVar
负责存储单一值,其所有操作都是同步的,所以可以通过它安全地访问各种可变对象。

其最主要的由三个方法:get()take()put(),其中 get()
take()
都负责从对象中获取值,且提供了无参版本和有超时时间的版本,区别是 get()
仅仅取出数据,而 take() 取出后会把原来的数据删除。

例:

val v = new SyncVar[Int]
Future {
  Thread.sleep(10)
  v.put(3)
}
var result = v.get
println("result", result, Thread.currentThread().getName) //(result,3,main)

result = v.take()
println("take", result, Thread.currentThread().getName) //(take,3,main)

//  以下代码由于值已经被取出所以会一直阻塞下去
v.get

Kotlin

Kotlin 基本就是直接调用 Java 代码,但是由于 Kotlin
对部分方法进行了包装,所以可以节省一定的代码量。

Thread

定义一个简单自动启动的线程

thread {
    println("async")
}
println("sync")

Kotlin 可以直接使用 Lambda 表达式生成
Thread,在创建时也可以像以下方法一样传入默认参数。

thread(start = false,
        isDaemon = true,
        name = "t1") {
    println("running")
}.start()

Thread Pool

创建一个单线程的线程池

val service = Executors.newSingleThreadExecutor()
service.execute {
    println("Running in a pool.")
}

Callable 与 Future

val future = service.submit(
        Callable() {
            "Hello World"
        }
)
val result = future.get()
println(result)
service.shutdown()

Timer

val fixedRateTimer = fixedRateTimer(name = "timer",
        daemon = true, initialDelay = 100, period = 100) {
    println("ping!")
}
Thread.sleep(1000)
fixedRateTimer.cancel()

CompletableFuture

val firstFuture = CompletableFuture<Int>()
firstFuture.complete(10)
println(firstFuture.get())

service = Executors.newFixedThreadPool(10)
val secondFuture = CompletableFuture.supplyAsync(Supplier {
    Random().nextInt(1000)
}, service)
println(secondFuture.get())
service.shutdown()

Summary

  • 不同语言的内部实现都是依赖 Java API,所以非常相似
  • Scala 的 Future 和 Promise 和 Java 1.8 引入的 CompletableFuture
    可以更好的写出异步代码

文章源码见
https://github.com/SidneyXu/JGSK
仓库的 _30_thread_future 小节

发表评论

电子邮件地址不会被公开。 必填项已用*标注