大家好，我是小黑，一個在互聯網茍且偷生的農民工，
Runnable
在創建執行緒時，可以通過new Thread(Runnable)方式，將任務代碼封裝在Runnable的run()方法中，將Runnable作為任務提交給Thread，或者使用執行緒池的execute(Runnable)方法處理，
public class RunnableDemo {
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
executorService.submit(new MyRunnable());
}
}
class MyRunnable implements Runnable {
@Override
public void run() {
System.out.println("runnable正在執行");
}
}
Runnable的問題
如果你之前有看過或者寫過Runnable相關的代碼，肯定會看到有說Runnable不能獲取任務執行結果的說法，這就是Runnable存在的問題，那么可不可以改造一下來滿足使用Runnable并獲取到任務的執行結果呢？答案是可以的，但是會比較麻煩，
首先我們不能修改run()方法讓它有回傳值，這違背了介面實作的原則；我們可以通過如下三步完成：
- 我們可以在自定義的
Runnable中定義變數，存盤計算結果； - 對外提供方法，讓外部可以通過方法獲取到結果；
- 在任務執行結束之前如果外部要獲取結果，則進行阻塞；
如果你有看過我之前的文章，相信要做到功能并不復雜，具體實作可以看我下面的代碼，
public class RunnableDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
MyRunnable<String> myRunnable = new MyRunnable<>();
new Thread(myRunnable).start();
System.out.println(LocalDateTime.now() + " myRunnable啟動~");
MyRunnable.Result<String> result = myRunnable.getResult();
System.out.println(LocalDateTime.now() + " " + result.getValue());
}
}
class MyRunnable<T> implements Runnable {
// 使用result作為回傳值的存盤變數，使用volatile修飾防止指令重排
private volatile Result<T> result;
@Override
public void run() {
// 因為在這個程序中會對result進行賦值，保證在賦值時外部執行緒不能獲取，所以加鎖
synchronized (this) {
try {
TimeUnit.SECONDS.sleep(2);
System.out.println(LocalDateTime.now() + " run方法正在執行");
result = new Result("這是回傳結果");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 賦值結束后喚醒等待執行緒
this.notifyAll();
}
}
}
// 方法加鎖，只能有一個執行緒獲取
public synchronized Result<T> getResult() throws InterruptedException {
// 回圈校驗是否已經給結果賦值
while (result == null) {
// 如果沒有賦值則等待
this.wait();
}
return result;
}
// 使用內部類包裝結果而不直接使用T作為回傳結果
// 可以支持回傳值等于null的情況
static class Result<T> {
T value;
public Result(T value) {
this.value = https://www.cnblogs.com/heiz123/archive/2021/09/16/value;
}
public T getValue() {
return value;
}
}
}
從運行結果我們可以看出，確實能夠在主執行緒中獲取到Runnable的回傳結果，
以上代碼看似從功能上可以滿足了我們的要求，但是存在很多并發情況的問題，實際開發中極不建議使用，在我們實際的作業場景中這樣的情況非常多，我們不能每次都這樣自定義搞一套，并且很容易出錯，造成執行緒安全問題，那么在JDK中已經給我們提供了專門的API來滿足我們的要求，它就是Callable，
Callable
我們通過Callable來完成我們上面說的1-1億的累加功能，
public class CallableDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Long max = 100_000_000L;
Long avgCount = max % 3 == 0 ? max / 3 : max / 3 + 1;
// 在FutureTask中存放結果
List<FutureTask<Long>> tasks = new ArrayList<>();
for (int i = 0; i < 3; i++) {
Long begin = 1 + avgCount * i;
Long end = 1 + avgCount * (i + 1);
if (end > max) {
end = max;
}
FutureTask<Long> task = new FutureTask<>(new MyCallable(begin, end));
tasks.add(task);
new Thread(task).start();
}
for (FutureTask<Long> task : tasks) {
// 從task中獲取任務處理結果
System.out.println(task.get());
}
}
}
class MyCallable implements Callable<Long> {
private final Long min;
private final Long max;
public MyCallable(Long min, Long max) {
this.min = min;
this.max = max;
}
@Override
public Long call() {
System.out.println("min:" + min + ",max:" + max);
Long sum = 0L;
for (Long i = min; i < max; i++) {
sum = sum + i;
}
// 可以回傳計算結果
return sum;
}
}
運行結果：
可以在創建執行緒時將Callable物件封裝在FutureTask物件中，交給Thread物件執行，
FutureTask之所以可以作為Thread創建的引數，是因為FutureTask是Runnable介面的一個實作類，
既然FutureTask也是Runnable介面的實作類，那一定也有run()方法，我們來通過原始碼看一下是怎么做到有回傳值的，
首先在FutureTask中有如下這些資訊，
public class FutureTask<V> implements RunnableFuture<V> {
// 任務的狀態
private volatile int state;
private static final int NEW = 0;
private static final int COMPLETING = 1;
private static final int NORMAL = 2;
private static final int EXCEPTIONAL = 3;
private static final int CANCELLED = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED = 6;
// 具體任務物件
private Callable<V> callable;
// 任務回傳結果或者例外時回傳的例外物件
private Object outcome;
// 當前正在運行的執行緒
private volatile Thread runner;
//
private volatile WaitNode waiters;
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
}
public void run() {
// 任務狀態的校驗
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
// 執行callable的call方法獲取結果
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
// 有例外則設定回傳值為ex
setException(ex);
}
// 執行程序沒有例外則將結果set
if (ran)
set(result);
}
} finally {
runner = null;
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
在這個方法中的核心邏輯就是執行callable的call()方法，將結果賦值，如果有例外則封裝例外，
然后我們看一下get方法如何獲取結果的，
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
// 這里會阻塞等待
s = awaitDone(false, 0L);
// 回傳結果
return report(s);
}
private V report(int s) throws ExecutionException {
Object x = outcome;
if (s == NORMAL)
return (V)x;
if (s >= CANCELLED)
// 狀態例外情況會拋出例外
throw new CancellationException();
throw new ExecutionException((Throwable)x);
}
在FutureTask中除了get()方法還提供有一些其他方法，
-
get(timeout,unit)：獲取結果，但只等待指定的時間；
-
cancel(boolean mayInterruptIfRunning)：取消當前任務；
-
isDone()：判斷任務是否已完成，
CompletableFuture
在使用FutureTask來完成異步任務，通過get()方法獲取結果時，會讓獲取結果的執行緒進入阻塞等待，這種方式并不是最理想的狀態，
在JDK8中引入了CompletableFuture，對Future進行了改進，可以在定義CompletableFuture傳入回呼物件，任務在完成或者例外時，自動回呼，
public class CompletableFutureDemo {
public static void main(String[] args) throws InterruptedException {
// 創建CompletableFuture時傳入Supplier物件
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new MySupplier());
//執行成功時
future.thenAccept(new MyConsumer());
// 執行例外時
future.exceptionally(new MyFunction());
// 主任務可以繼續處理，不用等任務執行完畢
System.out.println("主執行緒繼續執行");
Thread.sleep(5000);
System.out.println("主執行緒執行結束");
}
}
class MySupplier implements Supplier<Integer> {
@Override
public Integer get() {
try {
// 任務睡眠3s
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 3 + 2;
}
}
// 任務執行完成時回呼Consumer物件
class MyConsumer implements Consumer<Integer> {
@Override
public void accept(Integer integer) {
System.out.println("執行結果" + integer);
}
}
// 任務執行例外時回呼Function物件
class MyFunction implements Function<Throwable, Integer> {
@Override
public Integer apply(Throwable type) {
System.out.println("執行例外" + type);
return 0;
}
}
以上代碼可以通過lambda運算式進行簡化，
public class CompletableFutureDemo {
public static void main(String[] args) throws InterruptedException {
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
try {
// 任務睡眠3s
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 3 + 2;
});
//執行成功時
future.thenAccept((x) -> {
System.out.println("執行結果" + x);
});
future.exceptionally((type) -> {
System.out.println("執行例外" + type);
return 0;
});
System.out.println("主執行緒繼續執行");
Thread.sleep(5000);
System.out.println("主執行緒執行結束");
}
}
通過示例我們發現CompletableFuture的優點：
- 異步任務結束時，會自動回呼某個物件的方法；
- 異步任務出錯時，會自動回呼某個物件的方法；
- 主執行緒設定好回呼后，不再關心異步任務的執行，
當然這些優點還不足以體現CompletableFuture的強大，還有更厲害的功能，
串行執行
多個CompletableFuture可以串行執行，如第一個任務先進行查詢，第二個任務再進行更新
public class CompletableFutureDemo {
public static void main(String[] args) throws InterruptedException {
// 第一個任務
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 1234);
// 第二個任務
CompletableFuture<Integer> secondFuture = future.thenApplyAsync((num) -> {
System.out.println("num:" + num);
return num + 100;
});
secondFuture.thenAccept(System.out::println);
System.out.println("主執行緒繼續執行");
Thread.sleep(5000);
System.out.println("主執行緒執行結束");
}
}
并行任務
CompletableFuture除了可以串行，還支持并行處理，
public class CompletableFutureDemo {
public static void main(String[] args) throws InterruptedException {
// 第一個任務
CompletableFuture<Integer> oneFuture = CompletableFuture.supplyAsync(() -> 1234);
// 第二個任務
CompletableFuture<Integer> twoFuture = CompletableFuture.supplyAsync(() -> 5678);
// 通過anyOf將兩個任務合并為一個并行任務
CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(oneFuture, twoFuture);
anyFuture.thenAccept(System.out::println);
System.out.println("主執行緒繼續執行");
Thread.sleep(5000);
System.out.println("主執行緒執行結束");
}
}
通過anyOf()可以實作多個任務只有一個成功，CompletableFuture還有一個allOf()方法實作了多個任務必須都成功之后的合并任務，
小結
Runnable介面實作的異步執行緒默認不能回傳任務運行的結果，當然可以通過改造實作回傳，但是復雜度高，不適合進行改造；
Callable介面配合FutureTask可以滿足異步任務結果的回傳，但是存在一個問題，主執行緒在獲取不到結果時會阻塞等待；
CompletableFuture進行了增強，只需要指定任務執行結束或例外時的回呼物件，在結束后會自動執行，并且支持任務的串行，并行和多個任務都執行完畢后再執行等高級方法，
以上就是本期的全部內容，我們下期見，如果覺得有用點個關注唄，
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/300603.html
標籤：其他