Como esperar por um número de threads para completar?

Question

O que é uma maneira de simplesmente esperar por todo o processo de rosca ao fim? Por exemplo, digamos que eu tenho:

public class DoSomethingInAThread implements Runnable{

    public static void main(String[] args) {
        for (int n=0; n<1000; n++) {
            Thread t = new Thread(new DoSomethingInAThread());
            t.start();
        }
        // wait for all threads' run() methods to complete before continuing
    }

    public void run() {
        // do something here
    }


}

Como posso alterar isso para as pausas método main() no comentário até métodos run() saída de todos os tópicos? Obrigado!

Answer 1

Você coloca os tópicos em uma matriz, iniciá-los todos e, em seguida, tem um loop

for(i = 0; i < threads.length; i++)
  threads[i].join();

Cada juntar irá bloquear até que a respectiva discussão concluída. Threads podem completar em uma ordem diferente do que você se juntar a eles, mas isso não é um problema:., Quando o loop termina, os tópicos são concluídas

Answer 2

Uma forma seria a de fazer uma List de Threads, criar e lançar cada thread, acrescentando-o à lista. Uma vez que tudo é lançado, loop de volta a lista e chamada join() em cada um. Não importa que ordem os tópicos concluir a execução em, tudo que você precisa saber é que no momento em que termina segundo loop execução, cada segmento terá concluído.

Uma abordagem melhor é usar um ExecutorService e seus métodos associados:

List<Callable> callables = ... // assemble list of Callables here
                               // Like Runnable but can return a value
ExecutorService execSvc = Executors.newCachedThreadPool();
List<Future<?>> results = execSvc.invokeAll(callables);
// Note: You may not care about the return values, in which case don't
//       bother saving them

Usando um ExecutorService (e todas as coisas novas a partir do Java 5 utilitários de simultaneidade ) é extremamente flexível, e o exemplo acima quase não toca a superfície.

Answer 3

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class DoSomethingInAThread implements Runnable
{
   public static void main(String[] args) throws ExecutionException, InterruptedException
   {
      //limit the number of actual threads
      int poolSize = 10;
      ExecutorService service = Executors.newFixedThreadPool(poolSize);
      List<Future<Runnable>> futures = new ArrayList<Future<Runnable>>();

      for (int n = 0; n < 1000; n++)
      {
         Future f = service.submit(new DoSomethingInAThread());
         futures.add(f);
      }

      // wait for all tasks to complete before continuing
      for (Future<Runnable> f : futures)
      {
         f.get();
      }

      //shut down the executor service so that this thread can exit
      service.shutdownNow();
   }

   public void run()
   {
      // do something here
   }
}

Answer 4

Evite a classe Thread completamente e, em vez utilizar as abstrações mais elevados previstos no java.util.concurrent

A classe ExecutorService fornece o método invokeAll que parece fazer exatamente o que você quer.

Answer 5

em vez de join(), que é uma API de idade, você pode usar CountDownLatch . Eu modifiquei o código abaixo para cumprir sua exigência.

import java.util.concurrent.*;
class DoSomethingInAThread implements Runnable{
    CountDownLatch latch;
    public DoSomethingInAThread(CountDownLatch latch){
        this.latch = latch;
    } 
    public void run() {
        try{
            System.out.println("Do some thing");
            latch.countDown();
        }catch(Exception err){
            err.printStackTrace();
        }
    }
}

public class CountDownLatchDemo {
    public static void main(String[] args) {
        try{
            CountDownLatch latch = new CountDownLatch(1000);
            for (int n=0; n<1000; n++) {
                Thread t = new Thread(new DoSomethingInAThread(latch));
                t.start();
            }
            latch.await();
            System.out.println("In Main thread after completion of 1000 threads");
        }catch(Exception err){
            err.printStackTrace();
        }
    }
}

Explicação :

1. CountDownLatch foi inicializado com dados de contagem 1000 como por sua exigência.

2. Cada DoSomethingInAThread segmento de trabalho irá diminuir o CountDownLatch, que foi passado no construtor.

3. thread principal CountDownLatchDemo await() até a contagem tornou-se zero. Uma vez que a contagem tornou-se zero, você vai ficar abaixo da linha de saída.

   In Main thread after completion of 1000 threads
   

Mais informações a partir da página de documentação do Oracle

public void await()
           throws InterruptedException

faz com que o segmento atual de esperar até que o trinco tem contado até zero, a menos que o fio é interrompido.

Consulte a questão SE relacionados para outras opções:

esperar até os tópicos terminar seu trabalho em java

Answer 6

Como Martin K sugeriu java.util.concurrent.CountDownLatch parece ser uma solução melhor para isso. Basta adicionar um exemplo para o mesmo

     public class CountDownLatchDemo
{

    public static void main (String[] args)
    {
        int noOfThreads = 5;
        // Declare the count down latch based on the number of threads you need
        // to wait on
        final CountDownLatch executionCompleted = new CountDownLatch(noOfThreads);
        for (int i = 0; i < noOfThreads; i++)
        {
            new Thread()
            {

                @Override
                public void run ()
                {

                    System.out.println("I am executed by :" + Thread.currentThread().getName());
                    try
                    {
                        // Dummy sleep
                        Thread.sleep(3000);
                        // One thread has completed its job
                        executionCompleted.countDown();
                    }
                    catch (InterruptedException e)
                    {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }

            }.start();
        }

        try
        {
            // Wait till the count down latch opens.In the given case till five
            // times countDown method is invoked
            executionCompleted.await();
            System.out.println("All over");
        }
        catch (InterruptedException e)
        {
            e.printStackTrace();
        }
    }

}

Answer 7

Considere o uso de java.util.concurrent.CountDownLatch. Exemplos em javadocs

Answer 8

Dependendo de suas necessidades, você pode também querer verificar as classes CountDownLatch e CyclicBarrier no pacote java.util.concurrent. Eles podem ser úteis se você deseja que os seus tópicos que esperar para o outro, ou se você quer mais controle de grão fino sobre a forma como seus segmentos executar (por exemplo, à espera de sua execução interna para outro segmento para definir um estado). Você também pode usar um CountDownLatch para sinalizar todos os seus tópicos para começar ao mesmo tempo, em vez de começar um por um como você iterate através de seu loop. Os docs API padrão tem um exemplo disso, mais usando outro CountDownLatch que esperar para os tópicos para completar a sua execução.

Answer 9

Se você fizer uma lista dos tópicos, você pode loop através deles e .join () contra cada um, e seu ciclo vai terminar quando todas as threads têm. Eu não tentei-lo embora.

http: // docs .oracle.com / JavaSE / 8 / docs / api / java / lang / Thread.html # join ()

Answer 10

Este seria um comentário, mas eu não posso fazer comentários ainda.

Martin K , estou curioso sobre como você usaria ThreadGroup. Você já fez isso antes?

Eu vejo que acima, você sugerir verificar activeCount - deixando de lado Martin v Löwis 's preocupação com polling lado por um momento, eu tem outra preocupação com a própria activeCount.

Aviso: Eu não tentei usando isso, então eu não sou nenhum especialista no assunto, mas de acordo com o javadocs , ele retorna um estimativa do número de threads ativas.

Pessoalmente, eu estaria relutante em tentar construir um sistema em uma estimativa. Você tem outro pensamento sobre como fazê-lo, ou estou interpretando mal o javadoc?

Answer 11

Criar o objeto de discussão dentro do primeiro loop for.

for (int i = 0; i < threads.length; i++) {
     threads[i] = new Thread(new Runnable() {
         public void run() {
             // some code to run in parallel
         }
     });
     threads[i].start();
 }

E, em seguida, de modo que todo mundo aqui está dizendo.

for(i = 0; i < threads.length; i++)
  threads[i].join();

Answer 12

Você pode fazê-lo com o objeto " ThreadGroup" e seu parâmetro activeCount :

Answer 13

Como uma alternativa para CountDownLatch você também pode usar CyclicBarrier por exemplo.

public class ThreadWaitEx {
    static CyclicBarrier barrier = new CyclicBarrier(100, new Runnable(){
        public void run(){
            System.out.println("clean up job after all tasks are done.");
        }
    });
    public static void main(String[] args) {
        for (int i = 0; i < 100; i++) {
            Thread t = new Thread(new MyCallable(barrier));
            t.start();
        }       
    }

}    

class MyCallable implements Runnable{
    private CyclicBarrier b = null;
    public MyCallable(CyclicBarrier b){
        this.b = b;
    }
    @Override
    public void run(){
        try {
            //do something
            System.out.println(Thread.currentThread().getName()+" is waiting for barrier after completing his job.");
            b.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }       
}

Para usar CyclicBarrier neste caso barrier.await () deve ser a última declaração ou seja, quando o fio é feito com o seu trabalho. CyclicBarrier pode ser utilizado novamente com o seu método de reposição (). Para citar javadocs:

A CyclicBarrier suporta um comando Runnable opcional que é executado uma vez por ponto de barreira, após o último fio no partido chega, mas antes de qualquer segmentos são liberados. Esta ação barreira é útil para atualizar-estado compartilhado antes de qualquer das partes continuar.

Answer 14

O join() não foi útil para mim. veja este exemplo em Kotlin:

    val timeInMillis = System.currentTimeMillis()
    ThreadUtils.startNewThread(Runnable {
        for (i in 1..5) {
            val t = Thread(Runnable {
                Thread.sleep(50)
                var a = i
                kotlin.io.println(Thread.currentThread().name + "|" + "a=$a")
                Thread.sleep(200)
                for (j in 1..5) {
                    a *= j
                    Thread.sleep(100)
                    kotlin.io.println(Thread.currentThread().name + "|" + "$a*$j=$a")
                }
                kotlin.io.println(Thread.currentThread().name + "|TaskDurationInMillis = " + (System.currentTimeMillis() - timeInMillis))
            })
            t.start()
        }
    })

O resultado:

Thread-5|a=5
Thread-1|a=1
Thread-3|a=3
Thread-2|a=2
Thread-4|a=4
Thread-2|2*1=2
Thread-3|3*1=3
Thread-1|1*1=1
Thread-5|5*1=5
Thread-4|4*1=4
Thread-1|2*2=2
Thread-5|10*2=10
Thread-3|6*2=6
Thread-4|8*2=8
Thread-2|4*2=4
Thread-3|18*3=18
Thread-1|6*3=6
Thread-5|30*3=30
Thread-2|12*3=12
Thread-4|24*3=24
Thread-4|96*4=96
Thread-2|48*4=48
Thread-5|120*4=120
Thread-1|24*4=24
Thread-3|72*4=72
Thread-5|600*5=600
Thread-4|480*5=480
Thread-3|360*5=360
Thread-1|120*5=120
Thread-2|240*5=240
Thread-1|TaskDurationInMillis = 765
Thread-3|TaskDurationInMillis = 765
Thread-4|TaskDurationInMillis = 765
Thread-5|TaskDurationInMillis = 765
Thread-2|TaskDurationInMillis = 765

Agora, deixe-me usar o join() para tópicos:

    val timeInMillis = System.currentTimeMillis()
    ThreadUtils.startNewThread(Runnable {
        for (i in 1..5) {
            val t = Thread(Runnable {
                Thread.sleep(50)
                var a = i
                kotlin.io.println(Thread.currentThread().name + "|" + "a=$a")
                Thread.sleep(200)
                for (j in 1..5) {
                    a *= j
                    Thread.sleep(100)
                    kotlin.io.println(Thread.currentThread().name + "|" + "$a*$j=$a")
                }
                kotlin.io.println(Thread.currentThread().name + "|TaskDurationInMillis = " + (System.currentTimeMillis() - timeInMillis))
            })
            t.start()
            t.join()
        }
    })

E o resultado:

Thread-1|a=1
Thread-1|1*1=1
Thread-1|2*2=2
Thread-1|6*3=6
Thread-1|24*4=24
Thread-1|120*5=120
Thread-1|TaskDurationInMillis = 815
Thread-2|a=2
Thread-2|2*1=2
Thread-2|4*2=4
Thread-2|12*3=12
Thread-2|48*4=48
Thread-2|240*5=240
Thread-2|TaskDurationInMillis = 1568
Thread-3|a=3
Thread-3|3*1=3
Thread-3|6*2=6
Thread-3|18*3=18
Thread-3|72*4=72
Thread-3|360*5=360
Thread-3|TaskDurationInMillis = 2323
Thread-4|a=4
Thread-4|4*1=4
Thread-4|8*2=8
Thread-4|24*3=24
Thread-4|96*4=96
Thread-4|480*5=480
Thread-4|TaskDurationInMillis = 3078
Thread-5|a=5
Thread-5|5*1=5
Thread-5|10*2=10
Thread-5|30*3=30
Thread-5|120*4=120
Thread-5|600*5=600
Thread-5|TaskDurationInMillis = 3833

Como fica claro quando usamos o join:

1. Os segmentos estão executando sequencialmente.
2. A primeira amostra leva 765 milissegundos, enquanto a segunda amostra leva 3833 milissegundos.

---

A nossa solução para evitar bloqueio de outros segmentos foi a criação de um ArrayList:

val threads = ArrayList<Thread>()

Agora, quando queremos começar um novo segmento que mais adicioná-lo à ArrayList:

addThreadToArray(
    ThreadUtils.startNewThread(Runnable {
        ...
    })
)

A função addThreadToArray:

@Synchronized
fun addThreadToArray(th: Thread) {
    threads.add(th)
}

O funstion startNewThread:

fun startNewThread(runnable: Runnable) : Thread {
    val th = Thread(runnable)
    th.isDaemon = false
    th.priority = Thread.MAX_PRIORITY
    th.start()
    return th
}

Verifique a conclusão dos tópicos como abaixo em todos os lugares que seja necessário:

val notAliveThreads = ArrayList<Thread>()
for (t in threads)
    if (!t.isAlive)
        notAliveThreads.add(t)
threads.removeAll(notAliveThreads)
if (threads.size == 0){
    // The size is 0 -> there is no alive threads.
}