I'm wondering whether there is any advantage to keeping the same threads over the course of the execution of an object, rather than re-using the same Thread objects. I have an object for which a single (frequently used) method is parallelized using local Thread variables, such that every time the method is called, new Threads (and Runnables) are instantiated. Because the method is called so frequently, a single execution may instantiate upwards of a hundred thousand Thread objects, even though there are never more than a few (~4-6) active at any given time.

Following is a cut down example of how this method is currently implemented, to give a sense of what I mean. For reference, n is of course the pre-determined number of threads to use, whereas this.dataStructure is a (thread-safe) Map which serves as the input to the computation, as well as being modified by the computation. There are other inputs involved, but as they are not relevant to this question, I've omitted their usage. I've also omitted exception handling for the same reason.

Runnable[] tasks = new Runnable[n];
Thread[] threads = new Thread[n];
ArrayBlockingQueue<MyObject> inputs = new ArrayBlockingQueue<>(this.dataStructure.size());
inputs.addAll(this.dataStructure.values());

for (int i = 0; i < n; i  ) {
    tasks[i] = () -> {
        while (true) {
            MyObject input = inputs.poll(1L, TimeUnit.MICROSECONDS);
            if (input == null) return;
            // run computations over this.dataStructure
        }
    };
    threads[i] = new Thread(tasks[i]);
    threads[i].start();
}

for (int i = 0; i < n; i  )
    threads[i].join();

Because these Threads (and their runnables) always execute the same way using a single ArrayBlockingQueue as input, an alternative to this would be to just "refill the queue" every time the method is called and just re-start the same Threads. This is easily implemented, but I'm unsure as to whether it would make any difference one way or the other. I'm not too familiar with concurrency, so any help is appreciated.

PS.: If there is a more elegant way to handle the polling, that would also be helpful.

CodePudding user response：

Per definition, you cannot restart a thread. According to the documentation: It is never legal to start a thread more than once. In particular, a thread may not be restarted once it has completed execution.

Nevertheless a thread is a valuable resource, and there are implementations to reuse threads. Have a look at the Java Tutorial about Executors.

CodePudding user response：

It is not possible to start a Thread more than once, but conceptually, the answer to your question is yes.

This is normally accomplished with a thread pool. A thread pool is a set of Threads which rarely actually terminate. Instead, an application is passes its task to the thread pool, which picks a Thread in which to run it. The thread pool then decides whether the Thread should be terminated or reused after the task completes.

Java has some classes which make use of thread pools quite easy: ExecutorService and CompletableFuture.

ExecutorService usage typically looks like this:

ExecutorService executor = Executors.newCachedThreadPool();

for (int i = 0; i < n; i  ) {
    tasks[i] = () -> {
        while (true) {
            MyObject input = inputs.poll(1L, TimeUnit.MICROSECONDS);
            if (input == null) return;
            // run computations over this.dataStructure
        }
    };
    executor.submit(tasks[i]);
}

// Doesn't interrupt or halt any tasks.  Will wait for them all to finish
// before terminating its threads.
executor.shutdown();

executor.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);

Executors has other methods which can create thread pools, like newFixedThreadPool() and newWorkStealingPool(). You can decide for yourself which one best suits your needs.

CompletableFuture use might look like this:

Runnable[] tasks = new Runnable[n];
CompletableFuture<?>[] futures = new CompletableFuture<?>[n];

for (int i = 0; i < n; i  ) {
    tasks[i] = () -> {
        while (true) {
            MyObject input = inputs.poll(1L, TimeUnit.MICROSECONDS);
            if (input == null) return;
            // run computations over this.dataStructure
        }
    };
    futures[i] = CompletableFuture.runAsync(tasks[i]);
}

CompletableFuture.allOf(futures).get();

The disadvantage of CompletableFuture is that the tasks cannot be canceled or interrupted. (Calling cancel will mark the task as completing with an exception instead of completing successfully, but the task will not be interrupted.)