Come si limitano i thread in ExecutorService?

Question

Uso ExecutorService per eseguire molte attività in thread diversi. A volte, troppe istanze eseguibili in attesa nel pool di thread possono causare il problema di memoria insufficiente.

Provo a scrivere un esecutore di lavori di blocco per risolverlo. C'è qualche soluzione ufficiale per farlo?

Ad esempio:

BlockingJobExecutor executor = new BlockingJobExecutor(3); 
    for (int i = 0; i < 1000; i++) { 
     executor.addJob(new Runnable() { 

      @Override 
      public void run() { 
       try { 
        Thread.sleep(1000); 
       } catch (InterruptedException e) { 
        e.printStackTrace(); 
       } 
       LogFactory.getLog(BTest.class).info("test " + System.currentTimeMillis()); 
      } 
     }); 
    } 
    executor.shutdown(); 

qui è la classe BlockingJobExecutor:

import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.TimeUnit; 
import java.util.concurrent.atomic.AtomicInteger; 

public class BlockingJobExecutor { 

    AtomicInteger counter = new AtomicInteger(); 
    ExecutorService service; 
    int threads; 

    public BlockingJobExecutor(int threads) { 
     if (threads < 1) { 
      throw new IllegalArgumentException("threads must be greater than 1."); 
     } 
     service = Executors.newFixedThreadPool(threads); 
     this.threads = threads; 
    } 

    static class JobWrapper implements Runnable { 
     BlockingJobExecutor executor; 
     Runnable job; 

     public JobWrapper(BlockingJobExecutor executor, Runnable job) throws InterruptedException { 
      synchronized (executor.counter) { 
       while (executor.counter.get() >= executor.limit()) { 
        executor.counter.wait(); 
       } 
      } 
      this.executor = executor; 
      this.job = job; 
     } 

     @Override 
     public void run() { 
      try { 
       job.run(); 
      } finally { 
       synchronized (executor.counter) { 
        executor.counter.decrementAndGet(); 
        executor.counter.notifyAll(); 
       } 
      } 
     } 
    } 

    public int limit() { 
     return threads; 
    } 

    public void shutdown() { 
     service.shutdown(); 
     try { 
      service.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); 
     } catch (InterruptedException e) { 
      throw new RuntimeException(e); 
     } 
    } 

    public void addJob(Runnable job) { 
     try { 
      service.execute(new JobWrapper(this, job)); 
     } catch (InterruptedException e) { 
      throw new RuntimeException(e); 
     } 
    } 

} 

Answer 1

Ci sono due modi in cui questo può accadere. Puoi avere troppi runnable in coda in attesa di essere eseguiti, o troppi thread in esecuzione contemporaneamente. Se ci sono troppi lavori in coda, è possibile utilizzare una dimensione fissa BlockingQueue nello ExecutorService per limitare il numero di elementi che possono essere messi in coda. Quindi, quando si tenta di accodare una nuova attività, l'operazione verrà bloccata finché non ci sarà spazio nella coda.

Se ci sono troppi thread in esecuzione in una volta, è possibile limitare quanti thread sono disponibili per eseguire attività in ExecutorService chiamando Executors.newFixedThreadPool con il numero di thread che si desidera.