Comment utiliser Java multithread Future pour obtenir des tâches asynchrones-javaDidacticiel-php.cn

Limitations de Runnable

Dans l'article précédent, nous en avons parlé en implémentant l'interface Runnable via le codage, vous obtiendrez une "tâche" limite qui s'exécute dans un thread spécifié (ou un pool de threads).

En observant à nouveau l'interface, il n'est pas difficile de constater qu'elle n'a pas de valeur de retour de méthode :

public interface Runnable {
    void run();
}

Copier après la connexion

Avant JDK1.5, si vous souhaitez utiliser les résultats d'exécution de la tâche, vous devez utiliser soigneusement le fil de discussion pour accéder aux ressources de la section critique. Utiliser Callback pour le découplage est un très bon choix. 回调 进行解耦是非常不错的选择。

练手小Demo -- 回顾既往文章知识

注意，为了减少篇幅使用了lambda，但jdk1.5之前并不支持lambda

将计算任务分离到其他线程执行，再回到主线程消费结果

我们将计算、IO等耗时任务丢到其他线程，让主线程专注于自身业务，假想它在接受用户输入以及处理反馈,但我们略去这一部分

我们可以设计出类似下面的代码：

虽然它还有很多不合理之处值得优化，但也足以用于演示

class Demo {
    static final Object queueLock = new Object();
    static List<Runnable> mainQueue = new ArrayList<>();
    static boolean running = true;
    static final Runnable FINISH = () -> running = false;
    public static void main(String[] args) {
        synchronized (queueLock) {
            mainQueue.add(Demo::onStart);
        }
        while (running) {
            Runnable runnable = null;
            synchronized (queueLock) {
                if (!mainQueue.isEmpty())
                    runnable = mainQueue.remove(0);
            }
            if (runnable != null) {
                runnable.run();
            }
            Thread.yield();
        }
    }
    public static void onStart() {
        //...
    }
    public static void finish() {
        synchronized (queueLock) {
            mainQueue.clear();
            mainQueue.add(FINISH);
        }
    }
}

Copier après la connexion

再模拟一个计算的线程和任务回调：

interface Callback {
    void onResultCalculated(int result);
}
class CalcThread extends Thread {
    private final Callback callback;
    private final int a;
    private final int b;
    public CalcThread(Callback callback, int a, int b) {
        this.callback = callback;
        this.a = a;
        this.b = b;
    }
    @Override
    public void run() {
        super.run();
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        final int result = a + b;
        System.out.println("threadId" + Thread.currentThread().getId() + ",calc result:" + result + ";" + System.currentTimeMillis());
        synchronized (queueLock) {
            mainQueue.add(() -> callback.onResultCalculated(result));
        }
    }
}

Copier après la connexion

填充一下onStart业务：

class Demo {
    public static void onStart() {
        System.out.println("threadId" + Thread.currentThread().getId() + ",onStart," + System.currentTimeMillis());
        new CalcThread(result -> {
            System.out.println("threadId" + Thread.currentThread().getId() + ",onResultCalculated:" + result + ";" + System.currentTimeMillis());
            finish();
        }, 200, 300).start();
    }
}

Copier après la connexion

复习：优化为使用Runnable

在前文我们提到，如果业务仅关注任务的执行，并不过于关心线程本身，则可以利用Runnable：

class Demo {
    static class CalcRunnable implements Runnable {
        private final Callback callback;
        private final int a;
        private final int b;
        public CalcRunnable(Callback callback, int a, int b) {
            this.callback = callback;
            this.a = a;
            this.b = b;
        }
        @Override
        public void run() {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            final int result = a + b;
            System.out.println("threadId" + Thread.currentThread().getId() + ",calc result:" + result + ";" + System.currentTimeMillis());
            synchronized (queueLock) {
                mainQueue.add(() -> callback.onResultCalculated(result));
            }
        }
    }
    public static void onStart() {
        System.out.println("threadId" + Thread.currentThread().getId() + ",onStart," + System.currentTimeMillis());
        new Thread(new CalcRunnable(result -> {
            System.out.println("threadId" + Thread.currentThread().getId() + ",onResultCalculated:" + result + ";" + System.currentTimeMillis());
            finish();
        }, 200, 300)).start();
    }
}

Copier après la connexion

不难想象出：我们非常需要

让特定线程、特定类型的线程方便地接收任务，回顾本系列文章中的线程池篇，线程池是应运而生
拥有比Synchronize更轻量的机制
拥有更方便的数据结构

至此，我们可以体会到：JDK1.5之前，因为JDK的功能不足，Java程序对于线程的使用 较为粗糙。

为异步而生的Future

终于在JDK1.5中，迎来了新特性： Future 以及先前文章中提到的线程池, 时光荏苒，一晃将近20年了。

/**
 * 略
 * @since 1.5
 * @author Doug Lea
 * @param <V> The result type returned by this Future&#39;s {@code get} method
 */
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;
}

Copier après la connexion

尽管已经移除了API注释，但仍然能够理解每个API的含义，不多做赘述。

显而易见，为了增加返回值，没有必要用如此复杂的 接口来替代 Runnable。简单思考后可以对返回值的情况进行归纳：

返回Runnable中业务的结果，例如计算、读取资源等
单纯的在Runnable执行完毕后返回一个结果

从业务层上看，仅需要如下接口即可，它增加了返回值、并可以更友好地让使用者处理异常：

作者按：抛开底层实现，仅看业务方编码需要

public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

Copier après la connexion

显然，JDK需要提供后向兼容能力：

Runnable 不能够丢弃，也不应当丢弃
不能要求使用者完全的重构代码

所以一并提供了适配器，让使用者进行简单的局部重构即可用上新特性

static final class RunnableAdapter<T> implements Callable<T> {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

Copier après la connexion

而Future恰如其名，它代表了在 "未来" 的一个结果和状态，为了更方便地处理异步而生。

并且内置了 FutureTask，在 FutureTask详解章节中再行展开。

类图

在JDK1.8的基础上，看一下精简的类图结构：

Comment utiliser Java multithread Future pour obtenir des tâches asynchrones

FutureTask详解

构造函数

public class FutureTask {
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }
}

Copier après la connexion

生命周期

public class FutureTask {
    //新建
    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;
}

Copier après la connexion

可能的生命周期转换如下：

NEW -> COMPLETING -> NORMAL
NEW -> COMPLETING -> EXCEPTIONAL
NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED

JDK中原汁原味的解释如下：

The run state of this task, initially NEW. The run state transitions to a terminal state only in methods set, setException, and cancel. During completion, state may take on transient values of COMPLETING (while outcome is being set) or INTERRUPTING (only while interrupting the runner to satisfy a cancel(true)). Transitions from these intermediate to final states use cheaper ordered/lazy writes because values are unique and cannot be further modified.

核心方法

本节从以下三块入手阅读源码

状态判断
取消
获取结果

状态判断API的实现非常简单

public class FutureTask {
    public boolean isCancelled() {
        return state >= CANCELLED;
    }
    public boolean isDone() {
        return state != NEW;
    }
}

Copier après la connexion

取消：

当前状态为 NEW 且 CAS修改 state 成功，否则返回取消失败
如果 mayInterruptIfRunning
Petite démo pour s'entraîner - revoir les connaissances des articles précédents
Notez que lambda est utilisé pour réduire la longueur, mais jdk1.5 ne prend pas en charge lambda avant jdk1.5

Séparez les tâches de calcul des autres threads pour l'exécution, puis revenons au thread principal qui consomme les résultats

Nous confions des tâches fastidieuses telles que le calcul et les E/S à d'autres threads, permettant au thread principal de se concentrer sur ses propres activités. On suppose qu'il accepte l'utilisateur. saisie et traitement des commentaires, mais nous ignorons cette partie

Nous pouvons concevoir un code similaire à celui-ci :

Bien qu'il contienne encore de nombreuses choses déraisonnables qui méritent une optimisation, c'est suffisant pour la démonstration

public class FutureTask {
    public boolean cancel(boolean mayInterruptIfRunning) {
        if (!(state == NEW &&
                UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                        mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) {
            return false;
        }
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            finishCompletion();
        }
        return true;
    }
    private void finishCompletion() {
        // assert state > COMPLETING;
        for (WaitNode q; (q = waiters) != null; ) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (; ; ) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }
        done();
        callable = null;        // to reduce footprint
    }
}

Copier après la connexion

Simulez un fil de calcul et un rappel de tâche : 🎜

public class FutureTask {
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }
    public V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
        if (unit == null)
            throw new NullPointerException();
        int s = state;
        if (s <= COMPLETING &&
                (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
            throw new TimeoutException();
        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);
    }
}

Copier après la connexion

🎜Remplissez l'entreprise onStart : 🎜

class Demo {
   static class CalcResult {
      public int result;
   }
   public static void onStart() {
      System.out.println("threadId" + Thread.currentThread().getId() + ",onStart," + System.currentTimeMillis());
      final CalcResult calcResult = new CalcResult();
      Future<CalcResult> resultFuture = Executors.newSingleThreadExecutor().submit(() -> {
         try {
            Thread.sleep(10);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         final int result = 200 + 300;
         System.out.println("threadId" + Thread.currentThread().getId() + ",calc result:" + result + ";" + System.currentTimeMillis());
         calcResult.result = result;
      }, calcResult);
      System.out.println("threadId" + Thread.currentThread().getId() + "反正干点什么," + System.currentTimeMillis());
      if (resultFuture.isDone()) {
         try {
            final int ret = resultFuture.get().result;
            System.out.println("threadId" + Thread.currentThread().getId() + ",get result:" + ret + ";" + System.currentTimeMillis());
         } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
         }
      }
      finish();
   }
}

Copier après la connexion

Avis : Optimiser pour utiliser Runnable

🎜Nous avons mentionné dans l'article précédent que si l'entreprise se concentre uniquement sur l'exécution des tâches et ne se soucie pas trop de la thread lui-même, vous pouvez utiliser Runnable : 🎜

class Demo {
   public static void onStart() {
      System.out.println("threadId" + Thread.currentThread().getId() + ",onStart," + System.currentTimeMillis());
      ExecutorService executor = Executors.newSingleThreadExecutor();
      Future<CalcResult> resultFuture = executor.submit(() -> {
         try {
            Thread.sleep(10);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         final int result = 200 + 300;
         System.out.println("threadId" + Thread.currentThread().getId() + ",calc result:" + result + ";" + System.currentTimeMillis());
         final CalcResult calcResult = new CalcResult();
         calcResult.result = result;
         return calcResult;
      });
      System.out.println("threadId" + Thread.currentThread().getId() + "反正干点什么," + System.currentTimeMillis());
      if (resultFuture.isDone()) {
         try {
            final int ret = resultFuture.get().result;
            System.out.println("threadId" + Thread.currentThread().getId() + ",get result:" + ret + ";" + System.currentTimeMillis());
         } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
         }
      }
      executor.shutdown();
      finish();
   }
}

Copier après la connexion

🎜Ce n'est pas difficile à imaginer : nous en avons vraiment besoin🎜

🎜🎜 permet à des threads spécifiques et à des types spécifiques de threads de recevoir facilement des tâches . En regardant le pool de threads dans cette série d'articles, le pool de threads est né 🎜🎜🎜🎜 a plus d'avantages que Synchroniser Un mécanisme plus léger🎜🎜🎜🎜a une structure de données plus pratique🎜🎜🎜À ce stade. Nous pouvons nous en rendre compte : avant JDK1.5, en raison des fonctions insuffisantes du JDK, l'utilisation de threads dans les programmes Java étaitdifficile comparable. 🎜🎜Future, né pour l'asynchronisme🎜🎜, a enfin inauguré de nouvelles fonctionnalités dans JDK1.5 : Future et le pool de threads mentionnés dans l'article précédent. Le temps passe vite, près de 20 ans se sont écoulés. 🎜rrreee🎜Bien que les commentaires sur l'API aient été supprimés, vous pouvez toujours comprendre la signification de chaque API sans entrer dans les détails. 🎜🎜Évidemment, afin d'augmenter la valeur de retour, il n'est pas nécessaire de remplacer Runnable par une interface aussi complexe. Après une réflexion simple, nous pouvons résumer la valeur de retour : 🎜

🎜🎜Renvoyer les résultats de l'entreprise dans Runnable, tels que le calcul, la lecture des ressources, etc.🎜🎜🎜🎜Exécuter simplement dans Runnable Une fois terminé, un résultat est renvoyé🎜🎜🎜Du point de vue de la couche métier, seule l'interface suivante est nécessaire, ce qui ajoute une valeur de retour et permet aux utilisateurs de gérer les exceptions de manière plus conviviale : 🎜🎜Note de l'auteur : Laissez la couche inférieure derrière La mise en œuvre dépend uniquement des besoins de codage du côté commercial🎜rrreee🎜Évidemment, JDK doit fournir une compatibilité ascendante :🎜

🎜🎜Runnable ne peut pas être ignoré et doit ne pas être jeté🎜🎜🎜🎜Les utilisateurs ne sont pas tenus de refactoriser complètement le code🎜🎜🎜Un adaptateur est donc fourni pour permettre aux utilisateurs d'utiliser de nouvelles fonctionnalités via une simple refactorisation partielle🎜rrreee🎜Et Future est tout comme son son nom l'indique, il représente " Un résultat et un statut du "futur", créé pour gérer de manière asynchrone plus facilement. 🎜🎜Et il intègre FutureTask, qui sera développé dans le chapitre d'explication détaillée de FutureTask. 🎜

Diagramme de classes

🎜Basé sur JDK1.8, jetez un œil à la structure simplifiée du diagramme de classes : 🎜🎜 Comment utiliser Java multithread Future pour obtenir des tâches asynchrones

Comment utiliser Java multithread Future pour obtenir des tâches asynchrones

🎜🎜Explication détaillée de FutureTask🎜

Constructeur

rrreee

Cycle de vie

rrreee🎜Peut-être La conversion du cycle de vie est la suivante : 🎜

🎜🎜NOUVEAU -> TERMINÉ -> NORMAL🎜🎜NOUVEAU -> 🎜🎜NOUVEAU -> ; ANNULÉ🎜🎜🎜🎜NOUVEAU -> INTERRUPTEUR -> INTERROMPU🎜🎜🎜L'explication originale dans JDK est la suivante : 🎜

🎜L'état d'exécution de cette tâche, initialement NOUVEAU . L'état d'exécution passe à un état terminal uniquement dans les méthodes set, setException et Cancel. Pendant l'exécution, l'état peut prendre des valeurs transitoires de COMPLETING (pendant la définition du résultat) ou d'INTERRUPTING (uniquement lors de l'interruption du coureur pour satisfaire un objectif). Cancel(true)). Les transitions de ces états intermédiaires aux états finaux utilisent des écritures ordonnées/paresseuses moins chères car les valeurs sont uniques et ne peuvent pas être modifiées davantage.🎜

Méthode de base

🎜Cette section commence par les trois blocs suivants pour lire le code source🎜

🎜🎜Jugement du statut🎜🎜🎜🎜Annuler🎜🎜🎜🎜Obtenir le résultat🎜🎜🎜La mise en œuvre du statut L'API de jugement est très simple🎜rrreee🎜Annuler :🎜

NOUVEAU

mayInterruptIfRunning

public class FutureTask {
    public boolean cancel(boolean mayInterruptIfRunning) {
        if (!(state == NEW &&
                UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                        mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) {
            return false;
        }
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            finishCompletion();
        }
        return true;
    }
    private void finishCompletion() {
        // assert state > COMPLETING;
        for (WaitNode q; (q = waiters) != null; ) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (; ; ) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }
        done();
        callable = null;        // to reduce footprint
    }
}

Copier après la connexion

获取结果：先判断状态，如果未进入到 COMPLETING（即为NEW状态），则阻塞等待状态改变，返回结果或抛出异常

public class FutureTask {
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }
    public V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
        if (unit == null)
            throw new NullPointerException();
        int s = state;
        if (s <= COMPLETING &&
                (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
            throw new TimeoutException();
        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);
    }
}

Copier après la connexion

如何使用

而使用则非常简单，也非常的朴素。

我们以文中的的例子进行改造:

沿用原Runnable逻辑
移除回调，增加 CalcResult
将 CalcResult 对象作为既定返回结果，Runnable中设置其属性

class Demo {
   static class CalcResult {
      public int result;
   }
   public static void onStart() {
      System.out.println("threadId" + Thread.currentThread().getId() + ",onStart," + System.currentTimeMillis());
      final CalcResult calcResult = new CalcResult();
      Future<CalcResult> resultFuture = Executors.newSingleThreadExecutor().submit(() -> {
         try {
            Thread.sleep(10);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         final int result = 200 + 300;
         System.out.println("threadId" + Thread.currentThread().getId() + ",calc result:" + result + ";" + System.currentTimeMillis());
         calcResult.result = result;
      }, calcResult);
      System.out.println("threadId" + Thread.currentThread().getId() + "反正干点什么," + System.currentTimeMillis());
      if (resultFuture.isDone()) {
         try {
            final int ret = resultFuture.get().result;
            System.out.println("threadId" + Thread.currentThread().getId() + ",get result:" + ret + ";" + System.currentTimeMillis());
         } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
         }
      }
      finish();
   }
}

Copier après la connexion

如果直接使用新特性Callback，则如下：

直接返回结果，当然也可以直接返回Integer，不再包裹一层

class Demo {
   public static void onStart() {
      System.out.println("threadId" + Thread.currentThread().getId() + ",onStart," + System.currentTimeMillis());
      ExecutorService executor = Executors.newSingleThreadExecutor();
      Future<CalcResult> resultFuture = executor.submit(() -> {
         try {
            Thread.sleep(10);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         final int result = 200 + 300;
         System.out.println("threadId" + Thread.currentThread().getId() + ",calc result:" + result + ";" + System.currentTimeMillis());
         final CalcResult calcResult = new CalcResult();
         calcResult.result = result;
         return calcResult;
      });
      System.out.println("threadId" + Thread.currentThread().getId() + "反正干点什么," + System.currentTimeMillis());
      if (resultFuture.isDone()) {
         try {
            final int ret = resultFuture.get().result;
            System.out.println("threadId" + Thread.currentThread().getId() + ",get result:" + ret + ";" + System.currentTimeMillis());
         } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
         }
      }
      executor.shutdown();
      finish();
   }
}

Copier après la connexion

相信读者诸君会有这样的疑惑：

为何使用Future比原先的回调看起来粗糙？

首先要明确一点：文中前段的回调Demo，虽然达成了既定目标，但效率并不高！！在当时计算很昂贵的背景下，并不会如此莽撞地使用！

而在JDK1.5开始，提供了大量内容支持多线程开发。考虑到篇幅，会在系列文章中逐步展开。

另外，FutureTask中的CAS与Happens-Before本篇中亦不做展开。

接下来，再做一些引申，简单看一看多线程业务模式。