Come implementare il threading semplice con un numero fisso di thread di lavoro

Question

Sto cercando il modo più semplice e diretto per implementare quanto segue:

• Il programma principale crea un'istanza di lavoratore thread per svolgere un'attività.
• Solo le attività n possono essere eseguite contemporaneamente.
• Quando viene raggiunto n , non più lavoratori vengono avviati fino al conteggio di i thread in esecuzione tornano sotto n .

Answer 1

Penso che Executors.newFixedThreadPool soddisfa le tue esigenze. Esistono diversi modi per utilizzare ExecutorService risultante, a seconda che si desideri restituire un risultato al thread principale o se l'attività è totalmente autonoma e se si dispone di una raccolta di attività da eseguire in anticipo, oppure se le attività sono in coda in risposta ad alcuni eventi.

  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);

In alternativa, se hai una nuova attività asincrona da eseguire in risposta ad un evento, probabilmente vuoi semplicemente usare il semplice metodo execute (Runnable) di ExecutorService.

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

Executors. newFixedThreadPool (int)

Executor executor = Executors.newFixedThreadPool(n);

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

executor.execute(runnable);

Answer 4

Utilizza il framework Executor; ovvero newFixedThreadPool (N)

Answer 5

1. Se la coda delle attività non sarà illimitata e le attività possono essere completate in intervalli di tempo più brevi, è possibile utilizzare Executors.newFixedThreadPool (n) ; come suggerito dagli esperti.

   L'unico inconveniente di questa soluzione è la dimensione della coda delle attività senza limiti. Non hai il controllo su di esso. L'enorme accumulo nella coda delle attività peggiorerà le prestazioni dell'applicazione e potrebbe causare memoria insufficiente in alcuni scenari.

2. Se si desidera utilizzare ExecutorService e abilitare il meccanismo che ruba il lavoro dove i thread di lavoro inattivi condividono il carico di lavoro dai thread di lavoro occupati rubando le attività nella coda delle attività. Restituirà il tipo di servizio Executor ForkJoinPool.

     

   public static ExecutorService newWorkStealingPool (int parallelism)

        

   Crea un pool di thread che mantiene thread sufficienti per supportare il livello di parallelismo specificato e può utilizzare più code per ridurre la contesa. Il livello di parallelismo corrisponde al numero massimo di thread attivamente impegnati o disponibili a impegnarsi nell'elaborazione delle attività. Il numero effettivo di thread può aumentare e ridursi in modo dinamico. Un pool di furto di lavoro non fornisce alcuna garanzia sull'ordine in cui vengono eseguite le attività inviate.

3. Preferisco ThreadPoolExecutor a causa della flessibilità nelle API per controllare molti paratmetri, che controllano l'esecuzione dell'attività di flusso.

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

nel tuo caso, imposta sia corePoolSize che maximumPoolSize come N . Qui puoi controllare le dimensioni della coda delle attività, definire la tua fabbrica di thread personalizzata e la tua politica di gestione dei rifiuti.

Dai un'occhiata alla relativa domanda SE per controllare dinamicamente le dimensioni del pool:

Dynamic Thread Pool

Answer 6

Se vuoi lanciare il tuo:

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();
        } 
    }
}

Where Worker è la tua classe che estende Thread. Ogni lavoratore chiamerà il metodo removeWorker di questa classe, passandosi come parametro, quando ha finito di fare la cosa.

Detto questo, il framework Executor sembra molto meglio.

Modifica: qualcuno si preoccupa di spiegare perché questo è così male, invece di sminuirlo?

Answer 7

Come altri hanno già detto, la soluzione migliore è creare un pool di thread con Executors classe:

Tuttavia, se vuoi creare il tuo, questo codice dovrebbe darti un'idea di come procedere. Fondamentalmente, aggiungi ogni nuovo thread a un gruppo di thread e assicurati di non avere mai più di N thread attivi nel gruppo:

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);
    }
}