Run 100 threads in parallel and run missing threads if some previous are finished

Question

For example I need to always run 100 threads to do some action. I have class which called ThreadsWorker which looks for threads count and runs missing threads if some previous are finished. So, this is the table which describes situation:

1 second: 100 threads
2 second: 92 threads (ThreadsWorker generates new 8 threads)
3 second: 100 theads
4 second: 72 threads (ThreadsWorker generates 28 threads)

And so on. My threads are anonymous calls (just new Thread(new Runnable(...)).start()) because I don't know how to correctly save them to Threads[] array because, while ThreadsWorker will save threads[i] = new Threads(), some threads may be finished and then there will be some collision with array indexes.

Because of anonymous calls I use threadsCount variable now and increment it in threads body beginning and decrements in threads body end (using synchronized). Okay, it works correctly and my single way is to use while() loop which checks if threadsCount == 0 when the progress is complete.

I think that this is C-style but not Java-way :) So, can you help me to do it in Java-way?

Solution 1

If your goal is simply to have 100 threads actively processing, I suggest looking at Java thread pools (and Executors more generally).

I'm unclear as to whether you want to keep all 100 threads going or wait for them all to finish. Your question references both (ThreadsWorker spawning 28 new threads, threadsCount==0) and they seem contradictory.

Solution 2

Put all the threads into an array or collection.

Then loop through the collection calling Thread.join() on each. When this loop completes, all threads are done.

ArrayList threads = new ArrayList();
for (int i = 0; i < 5; i++) {
  Thread t = new AweseomeThread();
  t.start();
  threads.add(t);
}

for (Thread t : threads) {
  t.join();
}

You'll need some exception handling too (such as InterruptedException). But, I'll leave that as an exercise for the reader... :)

Solution 3

http://download.oracle.com/javase/6/docs/api/java/util/concurrent/CountDownLatch.html

You can try class CountDownLatch jdk api

private CountDownLatch latch;
private static class SimpleThread extends Thread {
 public void run() {
  latch.countDown();
 }
}
public static void main(String[] args) {
 int threadcount = 10;
 latch = new CountDownLatch(threadcount);
 for (int i = 0; i < 10; i++) {
  Thread t = new SimpleThread();
  t.start();
 }
 // waiting threads all finished
 latch.await();
}

Fetch the thread count from attibute latch of the Main class

Solution 4

I believe you are trying to have ThreadWorker submit new threads for all thread that have been completed.

I'd use a BlockingQueue that threads (Your Runnable(s)) add to when complete. ThreadWorker will wait until a thread completes and then will start a new thread.

public class YourRunnable implements Runnable {
  private final BlockingQueue<YourRunnable> queue;
  public YourRunnable(BlockingQueue<YourRunnable> queue){
    this.queue = queue;
  }
  public void run{
      // Your Code...
      // Finished Processing
      queue.add(this);
  }
}
public class ThreadWorkder implements Runnable { 
  private final BlockingQueue<YourRunnable> queue;
  ThreadWorker(BlockingQueue<YourRunnable> queue){
    this.queue = queue;
  }
  public void run{
    while(queue.take()){
       (new Thread(new YourRunnable(queue))).start();
    }
  }
  // general main method
  public static void main(String [] args){
    BlockingQueue<YourRunnable> queue = new LinkedBlockingQueue<YourRunnable>();
    ThreadWorker worker = new ThreadWorker(queue);
    Thread(worker).start();
    for (int i = 0; i < 100; i++){
      (new Thread(new YourRunnable(queue))).start();
    }
  }
}

Solution 5

Use a collection instead of an array. As the threads are completed, have them remove themselves from the array. Something like this:

public class Foo {
  Vector<Thread> threads = new Vector<Thread>(); //Vector is threadsafe

  public ensureThreadCount(int count) {
    while (threads.size() < count) {
      Thread t = new AweseomeThread(threads);
      threads.add(t);
      t.start();
    }
  }
}

public class AwesomeThread {
  Collection threads;
  public AwesomeThread(Collection threads) {
    this.threads = threads;
  }

  public void run() {
    try {
      // do stuff
    } catch (Throwable t) {
    } finally {
      threads.remove(this);
    }
  }
}

Then, have your worker just call Foo.ensureThreadCount().

View more solutions

Comments

[squawknull]

I think that's for waiting for a given operation in another thread to finish, but the poster asked to wait for the thread to finish and return, right? Wouldn't Thread.join() be more appropriate in his case?

[store88]

CountDownLatch can help Main for waiting all the threads finished, and cache the threads count.

[Clark]

Yep, I know about join method and exceptions :) But I don't know how to correctly work with indexes and threads array in this situation. I mean if I'll leave some indexes empty because of ThreadsWorker will think that threadsCount = 15 but while he'll run&save threads descriptors some another threads will die and f.e. threads[71] and threads[75] will be with descriptos but threads[72] to threads[74] will have empty descriptors. So I'll have to pause all threads while ThreadsWorker will save indexes to array to avoid empty indexes?

[Clark]

Thanks, I read today about CountDownLatch but I think that this is so massive for my task (join is brilliant method for this but I have some problems with indexed array). And maybe not a bit correct. But thanks anyway!

[Clark]

> «help Main for waiting all the threads finished, and cache the threads count» O! Can you show sample? After that I'll think what is more good for my situation.

[squawknull]

I would probably suggest not trying to use an array, but use a collection instead. Now that you mention that and I've re-read your question, I misunderstood. I thought you just wanted to know when all threads were complete, but it seems you want to keep 100 threads alive. I have another idea... I'll submit as a separate answer though for clarity and in case I'm still misunderstanding your question.

[andersoj]

+1 for using concurrency utils for this rather than raw threads. See my answer here for a bit of boilerplate code. stackoverflow.com/questions/5115940/…

[Clark]

So if I'll use «newFixedThreadPool()» then I'll should use some Worker which will create new threads to this queue for pool infinitly?