Pool de threads Java prenant en charge le blocage de la production

De manière générale, la vitesse des tâches de production est supérieure à la vitesse de consommation. Un détail est la longueur de la file d'attente et la manière de faire correspondre la vitesse de production et de consommation.

Un modèle producteur-consommateur typique est le suivant :

L'utilisation de l'implémentation de Queue fournie par J.U.C dans un environnement simultané peut facilement assurer la production et la consommation de Thread. sécurité dans le processus. Ce qu'il faut noter ici, c'est que la file d'attente doit définir la capacité initiale pour empêcher le producteur de produire trop rapidement, provoquant une montée en flèche de la longueur de la file d'attente et finalement déclenchant OutOfMemory.

Pour la situation générale où la production est plus rapide que la consommation. Lorsque la file d'attente est pleine, nous ne voulons pas qu'aucune tâche soit ignorée ou non exécutée. À ce stade, le producteur peut attendre un moment avant de soumettre la tâche. Une meilleure approche consiste à bloquer le producteur dans la méthode de soumission de la tâche. , et attendez que la tâche soit soumise. Continuez à soumettre des tâches lorsque la file d'attente n'est pas pleine, afin d'éviter toute perte de temps d'inactivité. Le blocage est également très simple. BlockingQueue est conçu pour cela. ArrayBlockingQueue et LinkedBlockingQueue peuvent fournir des limites de capacité lors de la construction de LinkedBlockingQueue qui détermine la capacité après l'obtention de chaque verrou lorsque la file d'attente est réellement exploitée.

De plus, lorsque la file d'attente est vide, le consommateur ne peut pas obtenir la tâche et peut attendre un moment avant de l'obtenir. Une meilleure approche consiste à utiliser la méthode take de BlockingQueue pour bloquer et attendre, et quand il y en a. une tâche, cela peut être fait immédiatement. Pour obtenir l'exécution, il est recommandé d'appeler la méthode surchargée de take avec un paramètre timeout. Le thread se fermera après le timeout. De cette manière, lorsque le producteur aura effectivement arrêté de produire, le consommateur ne sera pas laissé attendre indéfiniment.

Ainsi, un modèle de production et de consommation efficace prenant en charge le blocage est mis en œuvre.

Attendez une minute, puisque J.U.C nous a aidé à implémenter le pool de threads, pourquoi avons-nous encore besoin d'utiliser cet ensemble de choses ? N'est-il pas plus pratique d'utiliser directement ExecutorService ?

Jetons un coup d'œil à la structure de base de ThreadPoolExecutor :

Comme vous pouvez le voir, dans ThreadPoolExecutor, les parties BlockingQueue et Consumer ont été implémentées pour nous , et Il existe de nombreux avantages à utiliser directement l'implémentation du pool de threads, tels que l'ajustement dynamique du nombre de threads.

Mais le problème est que même si vous spécifiez manuellement un BlockingQueue comme implémentation de file d'attente lors de la construction de ThreadPoolExecutor, en fait, lorsque la file d'attente est pleine, la méthode d'exécution ne bloquera pas car ThreadPoolExecutor La méthode d'offre non bloquante de BlockingQueue s'appelle :

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
        if (runState == RUNNING && workQueue.offer(command)) {
            if (runState != RUNNING || poolSize == 0)
                ensureQueuedTaskHandled(command);
        }
        else if (!addIfUnderMaximumPoolSize(command))
            reject(command); // is shutdown or saturated
    }
}

A ce moment, vous devez faire quelque chose pour obtenir un résultat : lorsque le producteur soumet la tâche et que la file d'attente est pleine, le producteur peut être bloqué. Attendez que la tâche soit consommée.

La clé est que dans un environnement concurrent, le producteur ne peut pas déterminer si la file d'attente est pleine et ThreadPoolExecutor.getQueue().size() ne peut pas être appelé pour déterminer si la file d'attente est pleine.

Dans l'implémentation du pool de threads, lorsque la file d'attente est pleine, le RejectedExecutionHandler transmis lors de la construction sera appelé pour rejeter le traitement de la tâche. L'implémentation par défaut est AbortPolicy, qui lève directement une RejectedExecutionException.

Plusieurs stratégies de rejet ne seront pas décrites ici. Celle qui est la plus proche de nos besoins est CallerRunsPolicy. Cette stratégie permettra au thread qui a soumis la tâche d'exécuter la tâche lorsque la file d'attente est pleine, ce qui équivaut à. laisser la production Le producteur fait temporairement le travail du consommateur, de sorte que même si le producteur n'est pas bloqué, la tâche soumise sera également suspendue.

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    /**
     * Creates a <tt>CallerRunsPolicy</tt>.
     */
    public CallerRunsPolicy() { }

    /**
     * Executes task r in the caller&#39;s thread, unless the executor
     * has been shut down, in which case the task is discarded.
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

Cependant, cette stratégie comporte également des dangers cachés. Lorsqu'il y a peu de producteurs, pendant la période où le producteur consomme des tâches, le consommateur peut avoir fini de consommer toutes les tâches, laissant la file d'attente vide. . Lorsque le producteur exécute La tâche de production ne peut être poursuivie qu'une fois la tâche terminée. Ce processus peut entraîner une famine du thread consommateur.

En référence à des idées similaires, le moyen le plus simple est de définir directement un RejectedExecutionHandler, et lorsque la file d'attente est pleine, d'appeler BlockingQueue.put pour implémenter le blocage du producteur :

new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                if (!executor.isShutdown()) {
                        try {
                                executor.getQueue().put(r);
                        } catch (InterruptedException e) {
                                // should not be interrupted
                        }
                }
        }
};

De cette façon, nous nous n'avons plus besoin de nous soucier de la logique de la file d'attente et du consommateur, nous devons uniquement nous concentrer sur la logique d'implémentation des threads producteurs et consommateurs, et simplement soumettre les tâches au pool de threads.

Par rapport à la conception originale, cette méthode peut réduire considérablement la quantité de code et éviter de nombreux problèmes dans des environnements concurrents. Bien sûr, vous pouvez également utiliser d'autres méthodes, comme utiliser un sémaphore pour limiter les entrées lors de la soumission, mais si vous souhaitez simplement que le producteur bloque, ce sera compliqué.

Pour plus d'articles liés aux pools de threads Java prenant en charge le blocage de la production, veuillez faire attention au site Web PHP chinois !