Detalles y ejemplo de código de ThreadPool en Java

Antecedentes técnicos de la tecnología de la piscina de hilos

En la programación orientada a objetos, la creación y destrucción de objetos son muy costosas en términos de tiempo, porque la creación de un objeto requiere obtener recursos de memoria u otros recursos adicionales. En Java, esto es aún más cierto, ya que el motor virtual intenta rastrear cada objeto para poder realizar la recolección de basura después de que el objeto se destruye.

Por lo tanto, una de las formas de aumentar la eficiencia del programa de servicio es reducir al máximo el número de creaciones y destrucciones de objetos, especialmente las creaciones y destrucciones de objetos que consumen muchos recursos. Cómo utilizar los objetos existentes para proporcionar servicios es un problema clave que debe ser resuelto, y en realidad, esto es la razón de la aparición de algunas tecnologías de 'recursos de piscina'.

Por ejemplo, muchos componentes comunes que se ven comúnmente en Android, como las bibliotecas de carga de imágenes, las bibliotecas de solicitudes de red, incluso el mecanismo de transmisión de mensajes de Android, cuando se utiliza Message.obtain(), es un objeto de la piscina de Message. Por lo tanto, este concepto es muy importante. La tecnología de piscina de hilos que se presentará en este artículo también sigue este pensamiento.

Ventajas de la piscina de hilos:

1.Reutiliza los hilos de la piscina de hilos, reduciendo el costo de rendimiento debido a la creación y destrucción de objetos;

2.Puede controlar eficazmente el número máximo de concurrencias de hilos, mejorar la utilización de los recursos del sistema, al mismo tiempo evitar la competencia excesiva de recursos y evitar el bloqueo;

3.Puede gestionar de manera multihilo lo simple, haciendo que el uso de hilos sea simple y eficiente.

La estructura de la piscina de hilos Executor

En java, las piscinas de hilos se implementan a través del framework Executor, que incluye las clases: Executor, Executors, ExecutorService, ThreadPoolExecutor, Callable y Future, FutureTask, entre otros.

Executor: la interfaz de todos los pools de hilos, tiene solo un método.

public interface Executor {  
 void execute(Runnable command);  
}

ExecutorService: agrega comportamiento a Executor, es la interfaz más directa de la clase de implementación de Executor.

Executors: proporciona una serie de métodos de fábrica para crear piscinas de hilos, los pools de hilos devueltos implementan la interfaz ExecutorService.

ThreadPoolExecutor: la implementación específica de la piscina de hilos, generalmente todos los tipos de piscinas de hilos utilizados se implementan basándose en esta clase. El método de construcción es el siguiente:

public ThreadPoolExecutor(int corePoolSize,
        int maximumPoolSize,
        long keepAliveTime,
        TimeUnit unit,
        BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}

corePoolSize: Es el número de hilos de núcleo de la piscina de hilos, el número de hilos que se ejecutan en la piscina de hilos nunca excederá corePoolSize, en el caso predeterminado puede mantenerse vivo siempre. Se puede configurar allowCoreThreadTimeOut como True, en este caso, el número de hilos de núcleo es 0, y en este momento, keepAliveTime controla el tiempo de expiración de todos los hilos.

maximumPoolSize: Es el número máximo de hilos permitidos en la piscina de hilos;

keepAliveTime: Se refiere al tiempo de expiración de un hilo inactivo;

unit: Es una enumeración que representa la unidad de keepAliveTime;

workQueue: Representa la cola de bloqueo de tareas BlockingQueue<Runnable>.

BlockingQueue: La cola de bloqueo (BlockingQueue) es una herramienta principal utilizada en java.util.concurrent para controlar la sincronización de hilos. Si la cola de bloqueo está vacía, la operación de extracción de elementos de la cola de bloqueo se bloqueará y entrará en el estado de espera hasta que la cola de bloqueo tenga elementos. Del mismo modo, si la cola de bloqueo está llena, cualquier intento de almacenar elementos en ella también se bloqueará y entrará en el estado de espera hasta que haya espacio en la cola de bloqueo para continuar la operación. La cola de bloqueo se utiliza comúnmente en escenarios de productores y consumidores, donde el productor es el hilo que agrega elementos a la cola y el consumidor es el hilo que extrae elementos de la cola. La cola de bloqueo es el contenedor donde el productor almacena elementos, y el consumidor también solo extrae elementos del contenedor. Las implementaciones específicas incluyen LinkedBlockingQueue, ArrayBlockingQueue, etc. Generalmente, el bloqueo y la activación se implementan internamente a través de Lock y Condition (aprendizaje y uso de Lock y Condition).

El proceso de trabajo del pool de hilos es el siguiente: }}

Al crear el pool de hilos, no hay hilos en él. La cola de tareas se transmite como parámetro. Sin embargo, incluso si hay tareas en la cola, el pool de hilos no las ejecutará de inmediato.

Cuando se llama al método execute() para agregar una tarea, el pool de hilos realiza las siguientes evaluaciones:

Si el número de hilos en ejecución es menor que corePoolSize, cree inmediatamente un hilo para ejecutar esta tarea;

Si el número de hilos en ejecución es mayor o igual que corePoolSize, coloque esta tarea en la cola;

Si en este momento la cola está llena y el número de hilos en ejecución es menor que maximumPoolSize, aún se debe crear un hilo no nuclear para ejecutar esta tarea de inmediato;

Si la cola está llena y el número de hilos en ejecución es mayor o igual que maximumPoolSize, el pool de hilos lanzará una excepción RejectExecutionException.

Cuando un hilo completa una tarea, tomará la siguiente tarea de la cola para ejecutarla.

Cuando un hilo no tiene nada que hacer, después de un tiempo determinado (keepAliveTime), el pool de hilos lo evaluará. Si el número de hilos en ejecución es mayor que corePoolSize, este hilo se detendrá. Por lo tanto, después de que todas las tareas del pool de hilos se hayan completado, se reducirá finalmente al tamaño de corePoolSize.

Creación y uso del pool de hilos

La creación del pool de hilos utiliza el método estático de la herramienta Executors, a continuación se presentan varios tipos comunes de pools de hilos.

SingleThreadExecutor: un hilo de fondo único (su cola de buffer es ilimitada)

public static ExecutorService newSingleThreadExecutor() {  
 return new FinalizableDelegatedExecutorService (
  new ThreadPoolExecutor(1, 1,         
  0L, TimeUnit.MILLISECONDS,         
  new LinkedBlockingQueue<Runnable>())); 
}

Crear un pool de hilos monothread. Este pool de hilos tiene solo un hilo de trabajo en ejecución, es decir, equivalente a la ejecución serial de todas las tareas. Si este único hilo termina debido a una excepción, se sustituirá por un nuevo hilo. Este pool de hilos garantiza que todas las tareas se ejecuten en el orden de presentación de las tareas.

FixedThreadPool: una cola de hilos que solo tiene hilos de trabajo centrales, de tamaño fijo (su cola de buffer es ilimitada).

public static ExecutorService newFixedThreadPool(int nThreads) {　　　　　　　　
　　　　　　　 return new ThreadPoolExecutor(nThreads, nThreads,　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　
　　　　　　　　　　　 0L, TimeUnit.MILLISECONDS,　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　
　　　　　　　　　　　 new LinkedBlockingQueue<Runnable>());　　　　
}
Crear una cola de hilos de tamaño fijo. Cada vez que se presente una tarea, se creará un hilo, hasta que el tamaño del grupo de hilos alcance el tamaño máximo. Una vez que el tamaño del grupo de hilos alcance el valor máximo, se mantendrá inmutable. Si algún hilo termina debido a una excepción de ejecución, la cola de hilos complementará un nuevo hilo.

CachedThreadPool: una cola de hilos ilimitada, que puede realizar el reciclaje automático de hilos.

public static ExecutorService newCachedThreadPool() {   
 return new ThreadPoolExecutor(0, Integer.MAX_VALUE,           
   60L, TimeUnit.SECONDS,          
   new SynchronousQueue<Runnable>());  
}

Si el tamaño del grupo de hilos excede el número de hilos necesarios para procesar las tareas, se reciclarán algunos hilos idle (60 segundos sin ejecutar tareas) de hilos, cuando aumenta el número de tareas, esta cola de hilos también puede agregar inteligentemente nuevos hilos para procesar tareas. Esta cola de hilos no hace límite al tamaño del grupo de hilos, el tamaño del grupo de hilos depende completamente del tamaño máximo de hilos que puede crear el sistema operativo (o JVM). SynchronousQueue es una cola con un buffer de1de la cola de bloqueo.

ScheduledThreadPool: una cola de hilos de trabajo fija, un grupo de hilos con un tamaño ilimitado. Esta cola de hilos admite la ejecución programada y periódica de tareas.

public static ExecutorService newScheduledThreadPool(int corePoolSize) {   
 return new ScheduledThreadPool(corePoolSize, 
    Integer.MAX_VALUE,             
    DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,             
    new DelayedWorkQueue()); 
}

Crear una piscina de hilos que ejecuta tareas cíclicas. Si está inactivo, la piscina de hilos no nuclear se reciclará en DEFAULT_KEEPALIVEMILLIS.

Hay dos métodos comunes de presentación de tareas en la piscina de hilos:

execute:

ExecutorService.execute(Runnable runable)；

submit:

FutureTask task = ExecutorService.submit(Runnable runnable);
FutureTask<T> task = ExecutorService.submit(Runnable runnable, T Result);

FutureTask<T> task = ExecutorService.submit(Callable<T> callable);

La implementación de submit(Callable callable), y submit(Runnable runnable) es similar.

public <T> Future<T> submit(Callable<T> task) {
 if (task == null) throw new NullPointerException();
 FutureTask<T> ftask = newTaskFor(task);
 execute(ftask);
 return ftask;
}

Se puede ver que submit inicia una tarea con resultado devuelto, que devuelve un objeto FutureTask, de modo que se puede obtener el resultado a través del método get(). submit también llama a execute(Runnable runable) en última instancia, submit solo encapsula el objeto Callable o Runnable en un objeto FutureTask, porque FutureTask es Runnable, por lo que se puede ejecutar en execute. Sobre cómo encapsular Callable y Runnable en FutureTask, véase el uso de Callable y Future, FutureTask.

El principio de implementación de la piscina de hilos

Si solo se habla del uso de la piscina de hilos, este blog no tiene mucho valor, no más que el proceso de familiarización con las API relacionadas con Executor. El proceso de implementación de la piscina de hilos no utiliza la palabra clave Synchronized, sino que utiliza Volatile, Lock y sincronización (bloqueo) de colas, clases relacionadas con Atomic, FutureTask, etc., porque las últimas tienen un rendimiento mejor. El proceso de comprensión puede aprender bien la idea de control de concurrencia en el código fuente.

Como se mencionó al principio, las ventajas de la piscina de hilos se pueden resumir en los siguientes tres puntos:

Reciclaje de hilos

Controlar el número máximo de concurrencias

Gestión de hilos

1.Proceso de reciclaje de hilos

Para entender el principio del reciclaje de hilos, primero debe entender el ciclo de vida del hilo.

En el ciclo de vida de un hilo, debe pasar por nuevo (New), preparado (Runnable), en ejecución (Running), bloqueado (Blocked) y muerto (Dead)5estado.

Thread crea un nuevo hilo mediante new, este proceso es inicializar algunos información de hilo, como nombre de hilo, id, grupo de hilo, etc., que se puede considerar solo un objeto común. Después de llamar al método start() de Thread, la máquina virtual Java crea un pila de llamadas y un contador de programa, y establece hasBeenStarted en true, después de lo cual llamar al método start generará una excepción.

Los hilos en este estado no han comenzado a ejecutarse, solo indican que el hilo puede ejecutarse. Respecto a cuándo comenzará a ejecutarse el hilo, depende del administrador de hilos de JVM. Cuando el hilo obtiene la CPU, se llama al método run(). No debes llamar tú mismo al método run() de Thread. Después, según la programación de CPU, se cambia entre preparado, en ejecución y bloqueado, hasta que finaliza el método run() o se detiene el hilo de otra manera, entrando en estado muerto.

Por lo tanto, el principio de implementación del reciclaje de hilos debe ser mantener el hilo en estado de vida (preparado, en ejecución o bloqueado). Ahora veamos cómo ThreadPoolExecutor implementa el reciclaje de hilos.

En la clase principal Worker del ThreadPoolExecutor se controla el reciclaje de hilos. Vamos a ver el código simplificado de la clase Worker, para facilitar la comprensión:

private final class Worker implements Runnable {
final Thread thread;
Runnable firstTask;
Worker(Runnable firstTask) {
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
Tarea ejecutable = w.firstTask;
w.firstTask = null;
while (task != null || (task = getTask()) != null){
task.run();
}
}

Worker是一个Runnable，同时拥有一个thread，这个thread就是要开启的线程，在新建Worker对象时同时新建一个Thread对象，同时将Worker自己作为参数传入TThread，这样当Thread的start()方法调用时，运行的实际上是Worker的run()方法，接着到runWorker()中,有个while循环，一直从getTask()里得到Runnable对象，顺序执行。getTask()又是怎么得到Runnable对象的呢？

依旧是简化后的代码：

private Runnable getTask() {
 if(一些特殊情况) {
  return null;
 }
Runnable r = workQueue.take();
return r;
}

这个workQueue就是初始化ThreadPoolExecutor时存放任务的BlockingQueue队列，这个队列里的存放的都是将要执行的Runnable任务。因为BlockingQueue是个阻塞队列，BlockingQueue.take()得到如果是空，则进入等待状态直到BlockingQueue有新的对象被加入时唤醒阻塞的线程。所以一般情况Thread的run()方法就不会结束,而是不断执行从workQueue里的Runnable任务，这就达到了线程复用的原理了。

2.控制最大并发数

那Runnable是什么时候放入workQueue？Worker又是什么时候创建，Worker里的Thread的又是什么时候调用start()开启新线程来执行Worker的run()方法的呢？有上面的分析看出Worker里的runWorker()执行任务时是一个接一个，串行进行的，那并发是怎么体现的呢？

很容易想到是在execute(Runnable runnable)时会做上面的一些任务。看下execute里是怎么做的。

execute:

Código simplificado

public void execute(Runnable command) {
 if (command == null)
  throw new NullPointerException();
int c = ctl.get();
// Número de hilos actuales < corePoolSize
if (workerCountOf(c) < corePoolSize) {
// Inicie directamente un nuevo hilo.
if (addWorker(command, true))
return;
c = ctl.get();
}
// Número de hilos activos >= corePoolSize
// runState es RUNNING y la cola no está llena
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// . Vuelva a verificar si es estado RUNNING
// . Estado NO RUNNING, elimine la tarea de workQueue y rechace
if (!isRunning(recheck) && remove(command))
reject(command);// Rechazo de tareas según la estrategia especificada por el pool de hilos
// Dos situaciones:
// 1. Rechazo de nuevas tareas en estado NO RUNNING
// 2. Falla al iniciar un nuevo hilo cuando la cola está llena (workCount > maximumPoolSize)
} else if (!addWorker(command, false))
reject(command);
}

addWorker:

Código simplificado

private boolean addWorker(Runnable firstTask, boolean core) {
int wc = workerCountOf(c);
if (wc >= (core63; corePoolSize : maximumPoolSize)) {
return false;
}
w = new Worker(firstTask);
final Thread t = w.thread;
t.start();
}

Siguiendo el código, analicemos la situación de agregar tareas en el proceso de trabajo del pool de hilos mencionado anteriormente:

* Si el número de hilos en ejecución es menor que corePoolSize, cree inmediatamente un hilo para ejecutar esta tarea;　　
* Si el número de hilos en ejecución es mayor o igual que corePoolSize, coloque esta tarea en la cola;
* Si en este momento la cola está llena y el número de hilos en ejecución es menor que maximumPoolSize, aún se debe crear un hilo no nuclear para ejecutar esta tarea de inmediato;
* Si la cola está llena y el número de hilos en ejecución es mayor o igual que maximumPoolSize, el pool de hilos lanzará una excepción RejectExecutionException.

Esta es la razón por la que Android AsyncTask lanza RejectExecutionException cuando se ejecutan en paralelo y se excede el número máximo de tareas, consulte la explicación del código fuente de AsyncTask de la última versión y el lado oscuro de AsyncTask.

A través de addWorker, si se crea con éxito una nueva hilo, se inicia la nueva hilo mediante start() y se asigna firstTask como la primera tarea que se ejecutará en el run() de este Worker.

Aunque cada tarea del Worker se procesa en serie, si se crean múltiples Worker, ya que comparten el mismo workQueue, se procesan en paralelo.

Por lo tanto, se controla el número máximo de concurrencia según corePoolSize y maximumPoolSize. El proceso puede representarse aproximadamente con la siguiente imagen.

La explicación y la imagen anterior pueden entender bien este proceso.

Si eres un desarrollador de Android y estás muy familiarizado con el principio de Handler, puede que te sientas muy familiarizado con esta imagen, algunos procesos y Handler, Looper, Meaasge son muy similares en su uso. Handler.send(Message) es equivalente a execute(Runnuble), la cola de Meaasge mantenida por Looper es equivalente a BlockingQueue, pero es necesario mantener esta cola mediante sincronización, el ciclo de loop() en Looper extrae Meaasge de la cola de Meaasge y el runWork() en Worker extrae Runnable de la BlockingQueue de manera similar.

3.Gestión de hilos

a través del pool de hilos se puede gestionar bien la reutilización de hilos, controlar el número de concurrencia y otros procesos de eliminación, ya se ha hablado de la reutilización de hilos y el control de la concurrencia, y el proceso de gestión de hilos también se ha entrelazado en él, también es fácil de entender.

en ThreadPoolExecutor hay una variable AtomicInteger ctl. A través de esta variable se almacenan dos contenido:

número total de hilas, estado en el que se encuentra cada hilo, y baja29almacenamiento de bits de número de hilera, alto3almacenamiento de bits de runState, mediante operaciones de bits para obtener diferentes valores.

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//obtener el estado de la hilera
private static int runStateOf(int c) {
return c & ~CAPACITY;
}
//Obtener la cantidad de Worker
private static int workerCountOf(int c) {
return c & CAPACITY;
}
// Determinar si el hilo está en ejecución
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}

Aquí se analiza el proceso de cierre del pool de hilos a través de shutdown y shutdownNow(). Primero, el pool de hilos tiene cinco estados para controlar la adición y ejecución de tareas. Se presentan los siguientes tres:

Estado RUNNING: El pool de hilos funciona normalmente, puede aceptar nuevas tareas y procesar las tareas en la cola;

Estado SHUTDOWN: No se aceptan nuevas tareas, pero se ejecutarán las tareas en la cola;

Estado STOP: No se aceptan nuevas tareas, no se procesan las tareas en la cola; el método shutdown establecerá el runState en SHUTDOWN, detendrá todos los hilos disponibles, y los hilos que están trabajando no se verán afectados, por lo que las tareas en la cola se ejecutarán.

El método shutdownNow establece el runState en STOP. La diferencia con el método shutdown, este método detendrá todos los hilos, por lo que las tareas en la cola no se ejecutarán.

Resumen
A través del análisis del código fuente de ThreadPoolExecutor, se tiene una comprensión general del proceso de creación del pool de hilos, la adición de tareas y la ejecución, familiarizarse con estos procesos facilita el uso del pool de hilos.

Y algunas lecciones aprendidas sobre el control de concurrencia y el uso del modelo de productor-consumidor para la tarea de procesamiento, que serán muy útiles para entender o resolver otros problemas relacionados en el futuro. Por ejemplo, el mecanismo de Handler en Android, y el uso de una BloqueQueue para procesar la cola de Mensager en Looper también es posible, esto es lo que se aprende leyendo el código fuente.

Aquí está la recopilación de información sobre el Pool de Hilos de Java, se continuará complementando información relevante, ¡gracias por el apoyo a este sitio!