Ejecutores de Java: ¿cómo puedo establecer la prioridad de la tarea?

Question

¿Existe la posibilidad de asignar prioridad a las tareas que ejecutan los ejecutores? He encontrado algunas declaraciones en JCIP sobre esto, pero no puedo encontrar ningún ejemplo y no encuentro nada relacionado con los documentos.

De JCIP:

una política de ejecución especifica el "qué, dónde, cuándo y cómo" de la ejecución de la tarea , incluyendo:

• ...
• En lo orden en que se deben ejecutar las tareas (FIFO, LIFO, orden de prioridad)?
• ...

UPD: Me di cuenta de que me pidió que no exactamente lo que quería preguntar. Lo que realmente quería es:

Cómo utilizar/emular establecimiento de prioridades hilos (es decir, lo que fue thread.setPriority()) con el marco ejecutores?

Answer 1

Actualmente las únicas implementaciones concretas de the Executor interface son the ThreadPoolExecutor y the ScheduledThreadpoolExecutor

En lugar de utilizar la clase de utilidad/fábrica Executors, se debe crear una instancia utilizando un constructor.

Puede pasar un BlockingQueue a los constructores del ThreadPoolExecutor.

Una de las implementaciones de BlockingQueue, the PriorityBlockingQueue le permite pasar un Comparador a un constructor, de esa manera le permite decidir el orden de ejecución.

Answer 2

que es posible utilizar ThreadPoolExecutor con cola de prioridad bloqueo How to implement PriorityBlockingQueue with ThreadPoolExecutor and custom tasks

Answer 3

Tenga en cuenta que setPriority (..) normalmente hace no trabajo bajo Linux. Consulte los siguientes enlaces para los detalles completos:

• http://www.javamex.com/tutorials/threads/priority_what.shtml
• http://www.akshaal.info/2008/04/javas-thread-priorities-in-linux.html

Answer 4

puede especificar un ThreadFactory en el ThreadPoolExecutor constructor (método o Executors fábrica). Esto le permite proporcionar hilos de una prioridad de hilo dada para el ejecutor.

Para obtener diferentes prioridades de subprocesos para diferentes trabajos, debe enviarlos a ejecutores con diferentes fábricas de subprocesos.

Answer 5

La idea aquí es usar un PriorityBlockingQueue en el ejecutor. Para esto:

• Cree un comparador que compare nuestros futuros.
• Crea un proxy para el futuro para mantener una prioridad.
• Anule 'newTaskFor' para envolver cada futuro en nuestro proxy.

primer lugar usted necesita para mantener la prioridad en su futuro:

class PriorityFuture<T> implements RunnableFuture<T> { 

    private RunnableFuture<T> src; 
    private int priority; 

    public PriorityFuture(RunnableFuture<T> other, int priority) { 
     this.src = other; 
     this.priority = priority; 
    } 

    public int getPriority() { 
     return priority; 
    } 

    public boolean cancel(boolean mayInterruptIfRunning) { 
     return src.cancel(mayInterruptIfRunning); 
    } 

    public boolean isCancelled() { 
     return src.isCancelled(); 
    } 

    public boolean isDone() { 
     return src.isDone(); 
    } 

    public T get() throws InterruptedException, ExecutionException { 
     return src.get(); 
    } 

    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { 
     return src.get(); 
    } 

    public void run() { 
     src.run(); 
    } 
} 

continuación, tiene que definir comparador que clasificar correctamente los futuros prioritarios:

class PriorityFutureComparator implements Comparator<Runnable> { 
    public int compare(Runnable o1, Runnable o2) { 
     if (o1 == null && o2 == null) 
      return 0; 
     else if (o1 == null) 
      return -1; 
     else if (o2 == null) 
      return 1; 
     else { 
      int p1 = ((PriorityFuture<?>) o1).getPriority(); 
      int p2 = ((PriorityFuture<?>) o2).getPriority(); 

      return p1 > p2 ? 1 : (p1 == p2 ? 0 : -1); 
     } 
    } 
} 

A continuación vamos a suponer que tenemos una trabajo largo como este:

class LenthyJob implements Callable<Long> { 
    private int priority; 

    public LenthyJob(int priority) { 
     this.priority = priority; 
    } 

    public Long call() throws Exception { 
     System.out.println("Executing: " + priority); 
     long num = 1000000; 
     for (int i = 0; i < 1000000; i++) { 
      num *= Math.random() * 1000; 
      num /= Math.random() * 1000; 
      if (num == 0) 
       num = 1000000; 
     } 
     return num; 
    } 

    public int getPriority() { 
     return priority; 
    } 
} 

A continuación, con el fin de ejecutar estos trabajos en prioridad el código se verá así:

public class TestPQ { 

    public static void main(String[] args) throws InterruptedException, ExecutionException { 
     int nThreads = 2; 
     int qInitialSize = 10; 

     ExecutorService exec = new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, 
       new PriorityBlockingQueue<Runnable>(qInitialSize, new PriorityFutureComparator())) { 

      protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { 
       RunnableFuture<T> newTaskFor = super.newTaskFor(callable); 
       return new PriorityFuture<T>(newTaskFor, ((LenthyJob) callable).getPriority()); 
      } 
     }; 

     for (int i = 0; i < 20; i++) { 
      int priority = (int) (Math.random() * 100); 
      System.out.println("Scheduling: " + priority); 
      LenthyJob job = new LenthyJob(priority); 
      exec.submit(job); 
     } 
    } 
} 

Esta es una gran cantidad de código, pero eso es casi la única manera esto se puede lograr.

En mi máquina es la salida como la siguiente:

Scheduling: 39 
Scheduling: 90 
Scheduling: 88 
Executing: 39 
Scheduling: 75 
Executing: 90 
Scheduling: 15 
Scheduling: 2 
Scheduling: 5 
Scheduling: 24 
Scheduling: 82 
Scheduling: 81 
Scheduling: 3 
Scheduling: 23 
Scheduling: 7 
Scheduling: 40 
Scheduling: 77 
Scheduling: 49 
Scheduling: 34 
Scheduling: 22 
Scheduling: 97 
Scheduling: 33 
Executing: 2 
Executing: 3 
Executing: 5 
Executing: 7 
Executing: 15 
Executing: 22 
Executing: 23 
Executing: 24 
Executing: 33 
Executing: 34 
Executing: 40 
Executing: 49 
Executing: 75 
Executing: 77 
Executing: 81 
Executing: 82 
Executing: 88 
Executing: 97 

Answer 6

Sólo quiero añadir mi granito de arena de la contribución a esta discusión. Implementé este ReorderingThreadPoolExecutor para un propósito muy específico, que es poder llevar explícitamente al frente del BlockingQueue del ejecutor (en este caso LinkedBlockingDeque) cuando quiera y sin tener que lidiar con las prioridades (lo que puede conducir a interbloqueos y , de todos modos, arreglado).

Estoy usando esto para administrar (dentro de una aplicación de Android) el caso donde tengo que descargar muchas imágenes que se muestran en una vista de lista larga. Cada vez que el usuario se desplaza rápidamente, la cola del ejecutor se inunda de solicitudes de descarga de imágenes: al mover las últimas en la parte superior de la cola, he logrado mejores desempeños al cargar las imágenes que están realmente en la pantalla, retrasando la descarga de los que probablemente necesitarán más tarde. Tenga en cuenta que utilizo una clave de mapa concurrente interna (que puede ser tan simple como la cadena URL de la imagen) para agregar las tareas al ejecutor para que pueda recuperarlas más tarde para el reordenamiento.

Hubo muchas otras formas de hacer lo mismo y tal vez es demasiado complicado, pero funciona bien y también Facebook en su SDK de Android está haciendo algo similar en su propia fila de hilos de trabajo.

Siéntase libre de echar un vistazo al código y darme sugerencias, está dentro de un proyecto Android, pero pelar unos registros y anotaciones haría la clase Java puro 6.

Answer 7

Usted puede implementar su propio ThreadFactory y establecer dentro ThreadPoolExecutor así:

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, numOfWorkerThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); 
threadPool.setThreadFactory(new OpJobThreadFactory(Thread.NORM_PRIORITY-2)); 

donde mi OpJobThreadFactory tiene el siguiente aspecto:

public final static class OpJobThreadFactory implements ThreadFactory { 
    private int priority; 
    private boolean daemon; 
    private final String namePrefix; 
    private static final AtomicInteger poolNumber = new AtomicInteger(1); 
    private final AtomicInteger threadNumber = new AtomicInteger(1); 

    public OpJobThreadFactory(int priority) { 
     this(priority, true); 
    } 

    public OpJobThreadFactory(int priority, boolean daemon) { 
     this.priority = priority; 
     this.daemon = daemon; 
     namePrefix = "jobpool-" +poolNumber.getAndIncrement() + "-thread-"; 
    } 

    @Override 
    public Thread newThread(Runnable r) { 
     Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement()); 
     t.setDaemon(daemon); 
     t.setPriority(priority); 
     return t; 
    } 
}