Cómo implementar subprocesos simples con un número fijo de subprocesos de trabajo

Question

Estoy buscando la forma más sencilla y directa de implementar lo siguiente:

• El programa principal instancias de los hilos de trabajadores para hacer una tarea.
• Solo n Las tareas se pueden ejecutar a la vez.
• Cuando n se alcanza, no se inician más trabajadores hasta que el recuento de hilos en ejecución vuelve a caer n.

Answer 1

Creo que Ejecutores.newFixedThreadPool se ajusta a sus requerimientos.Hay varias formas diferentes de utilizar el ExecutorService resultante, dependiendo de si desea que se devuelva un resultado al hilo principal, o si la tarea es totalmente independiente y si tiene una colección de tareas para realizar por adelantado, o si las tareas están en cola en respuesta a algún evento.

  Collection<YourTask> tasks = new ArrayList<YourTask>();
  YourTask yt1 = new YourTask();
  ...
  tasks.add(yt1);
  ...
  ExecutorService exec = Executors.newFixedThreadPool(5);
  List<Future<YourResultType>> results = exec.invokeAll(tasks);

Alternativamente, si tiene que realizar una nueva tarea asincrónica en respuesta a algún evento, probablemente solo desee utilizar la sencilla función de ExecutorService. execute(Runnable) método.

Answer 2

/* Get an executor service that will run a maximum of 5 threads at a time: */
ExecutorService exec = Executors.newFixedThreadPool(5);
/* For all the 100 tasks to be done altogether... */
for (int i = 0; i < 100; i++) {
    /* ...execute the task to run concurrently as a runnable: */
    exec.execute(new Runnable() {
        public void run() {
            /* do the work to be done in its own thread */
            System.out.println("Running in: " + Thread.currentThread());
        }
    });
}
/* Tell the executor that after these 100 steps above, we will be done: */
exec.shutdown();
try {
    /* The tasks are now running concurrently. We wait until all work is done, 
     * with a timeout of 50 seconds: */
    boolean b = exec.awaitTermination(50, TimeUnit.SECONDS);
    /* If the execution timed out, false is returned: */
    System.out.println("All done: " + b);
} catch (InterruptedException e) { e.printStackTrace(); }

Answer 3

Ejecutores.newFixedThreadPool(int)

Executor executor = Executors.newFixedThreadPool(n);

Runnable runnable = new Runnable() {
 public void run() {
  // do your thing here
 }
}

executor.execute(runnable);

Answer 4

Utilice el marco Ejecutor;a saber nuevoFixedThreadPool(N)

Answer 5

1. Si su cola de tareas no va a ser ilimitada y las tareas pueden completarse en intervalos de tiempo más cortos, puede usar Executors.newFixedThreadPool(n);como sugieren los expertos.

   El único inconveniente de esta solución es el tamaño ilimitado de la cola de tareas.No tienes control sobre ello.La enorme acumulación en la cola de tareas degradará el rendimiento de la aplicación y puede provocar falta de memoria en algunos escenarios.

2. Si quieres usar ExecutorService y habilitar work stealing Mecanismo donde los subprocesos de trabajo inactivos comparten la carga de trabajo de los subprocesos de trabajo ocupados robando tareas en la cola de tareas.Devolverá el tipo de Servicio Ejecutor ForkJoinPool.

   estática pública ExecutorService newWorkStealingPool(int paralelismo)

   Crea un grupo de subprocesos que mantiene suficientes subprocesos para admitir el nivel de paralelismo determinado y puede utilizar varias colas para reducir la contención.El nivel de paralelismo corresponde al número máximo de subprocesos que participan activamente o están disponibles para participar en el procesamiento de tareas.El número real de subprocesos puede crecer y reducirse dinámicamente.Un grupo de robo de trabajo no ofrece garantías sobre el orden en que se ejecutan las tareas enviadas.

3. yo prefiero ThreadPoolExecutor debido a la flexibilidad de las API para controlar muchos parámetros, lo que controla la ejecución de la tarea de flujo.

   ThreadPoolExecutor(int corePoolSize, 
                          int maximumPoolSize, 
                          long keepAliveTime, 
                          TimeUnit unit, 
                          BlockingQueue<Runnable> workQueue, 
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)
   

en tu caso, configura ambos corePoolSize and maximumPoolSize as N.Aquí puede controlar el tamaño de la cola de tareas, definir su propia fábrica de subprocesos personalizada y su política de manejo de rechazos.

Eche un vistazo a la pregunta SE relacionada para controlar dinámicamente el tamaño del grupo:

Grupo de subprocesos dinámicos

Answer 6

Si quieres rodar el tuyo propio:

private static final int MAX_WORKERS = n;
private List<Worker> workers = new ArrayList<Worker>(MAX_WORKERS);

private boolean roomLeft() {
    synchronized (workers) {
        return (workers.size() < MAX_WORKERS);
    }
}

private void addWorker() {
    synchronized (workers) {
        workers.add(new Worker(this));
    }
}

public void removeWorker(Worker worker) {
    synchronized (workers) {
        workers.remove(worker);
    }
}

public Example() {
    while (true) {
        if (roomLeft()) {
            addWorker();
        } 
    }
}

Donde Worker es tu clase que extiende Thread.Cada trabajador llamará al método removeWorker de esta clase, pasándose a sí mismo como un parámetro, cuando termine de hacer su trabajo.

Dicho esto, el marco Executor se ve mucho mejor.

Editar:¿Alguien quiere explicar por qué esto es tan malo, en lugar de simplemente reducirlo?

Answer 7

Como otros han mencionado aquí, lo mejor que puede hacer es crear un grupo de subprocesos con el Ejecutores clase:

Sin embargo, si desea crear el suyo propio, este código debería darle una idea de cómo proceder.Básicamente, simplemente agregue cada hilo nuevo a un grupo de hilos y asegúrese de que nunca tenga más de N hilos activos en el grupo:

Task[] tasks = getTasks(); // array of tasks to complete
ThreadGroup group = new ThreadGroup();
int i=0;
while( i<tasks.length || group.activeCount()>0 ) {
    if( group.activeCount()<N && i<tasks.length ) {
        new TaskThread(group, tasks[i]).start();
        i++;
    } else {
        Thread.sleep(100);
    }
}