Wie einfaches Einfädeln mit einer festen Anzahl von Worker-Threads implementieren

Question

Ich interessiere mich für die einfachste, direkteste Weg, um die folgenden zu implementieren:

• Das Hauptprogramm instanziiert Arbeiter Fäden, eine Aufgabe zu tun.
• Nur n Aufgaben können gleichzeitig ausgeführt werden.
• Wenn n erreicht ist, nicht mehr Arbeiter bis die Zählung gestartet von Fäden fällt wieder unter n ausgeführt wird.

Answer 1

Ich denke, dass Executors.newFixedThreadPool Ihren Anforderungen passt. Es gibt eine Reihe von verschiedenen Möglichkeiten, um die resultierende ExecutorService zu verwenden, je nachdem, ob Sie ein Ergebnis an den Hauptthread zurückgegeben, oder ob die Aufgabe ist völlig eigenständig, und ob Sie eine Sammlung von Aufgaben Front auszuführen, oder ob Aufgaben als Reaktion auf ein Ereignis der Warteschlange.

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

Alternativ, wenn Sie eine neue asynchrone Aufgabe haben als Reaktion auf ein Ereignis auszuführen, wahrscheinlich wollen Sie nur die ExecutorService der einfache execute(Runnable) Methode verwenden.

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

Testamentsvollstrecker. newFixedThreadPool (int)

Executor executor = Executors.newFixedThreadPool(n);

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

executor.execute(runnable);

Answer 4

Mit dem Executor Rahmen; nämlich newFixedThreadPool (N)

Answer 5

1. Wenn Sie Ihre Aufgabe Warteschlange ist nicht unbegrenzt sein werde und Aufgaben können in kürzeren Zeitabständen abgeschlossen haben, können Sie Executors.newFixedThreadPool(n) verwenden; wie schon sagt von Experten.

   Der einzige Nachteil bei dieser Lösung ist unbegrenzt Aufgabe Warteschlangengröße. Sie haben keine Kontrolle darüber. Die riesige Massenkarambolage in Task-Queue-Leistung von Anwendung verschlechtert und kann aus dem Speicher in einigen Szenarien führen.

2. Wenn Sie ExecutorService verwenden möchten, und aktivieren work stealing Mechanismus, bei dem Leerlaufarbeitsthreads die Arbeitsbelastung von stark befahrenen Worker-Threads gemeinsam von Aufgaben in der Aufgabenwarteschlange zu stehlen. Es wird wieder ForkJoinPool Art von Executor-Service.

     

   public static ExecutorService newWorkStealingPool (int Parallelität)

        

   Erstellt einen Thread-Pool, der genügende Fäden hält die gegebene Parallelität Ebene zu unterstützen, und mehrere Warteschlangen Anstoß zu reduzieren verwenden. Die Parallelität Ebene entspricht die maximale Anzahl von Threads aktiv beteiligt ist, oder zur Verfügung zu engagieren, Aufgabenverarbeitung. Die tatsächliche Anzahl der Threads dynamisch wachsen und schrumpfen. Ein Work-Diebstahl-Pool übernimmt keine Garantie über die Reihenfolge, in der eingereichten Aufgaben ausgeführt werden.

3. Ich ziehe es durch Flexibilität in APIs ThreadPoolExecutor zu viele paratmeters zu steuern, die den Fluss der Aufgabenausführung steuert.

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

in Ihrem Fall, setzen beide corePoolSize and maximumPoolSize as N. Hier können Sie Task-Queue-Größe steuern, definieren Ihre eigenen Thread Fabrik und Ablehnung Handler Politik.

Hier finden Sie aktuelle relevante Frage SE die Poolgröße zu steuern, dynamisch:

Dynamische Thread Pool

Answer 6

Wenn Sie möchten Ihre eigene Rolle:

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

Wo Arbeiter ist Ihre Klasse, die Thread erweitert. Jeder Arbeiter wird Klasse dieser Aufruf removeWorker Methode, sich vorbei als Parameter, wenn es fertig ist zu tun, es ist Sache.

Mit dieser sagte, der Executor Rahmen sieht viel besser aus.

Edit:? Wer kümmert zu erklären, warum das so schlecht ist, anstatt es nur downmodding

Answer 7

Wie andere hier schon erwähnt haben, die beste Wahl ist, einen Thread-Pool mit dem Pfänder Klasse:

Wenn Sie jedoch Ihre eigene Rolle möchten, sollten Sie diesen Code geben Sie eine Idee, wie Sie vorgehen. Grundsätzlich fügen Sie einfach jeden neuen Thread zu einer Thread-Gruppe und stellen Sie sicher, dass Sie nie mehr als N aktive Threads in der Gruppe haben:

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