多线程列表操作

Question

多线程能同时处理来自同一个列表的值吗？例如，我们有3个线程和它们必须执行的任务列表：

ExecutorService executor = Executors.newFixedThreadPool(3);
List<Strng> tasks = Arrays.asList("firstTask", "secondTask", "thirdTask", "foiuthTask", "fifthTask");

线程开始工作并接受前三个任务，但第二个线程最快地完成了任务。现在它需要第四个任务来工作，如果它没有被另一个线程占用等等。这个实现是可能的吗？我找不到关于这件事的任何信息。

我试着使用并行流，但总是得到错误的结果。

Answer 1

您应该将任务提交给ExecutorService，它将处理分配给任何可用线程的任务。

executor.submit(task)

如果要在执行器之外执行此操作，还可以使用BlockingQueue而不是列表，因为这是ExecutorService在内部工作的方式。

Answer 2

我会将每个输入字符串传递给Runnable任务的一个实例。然后将任务提供给executor服务。

注意，Runnable没有要返回的结果。一个Callable就有。这里的示例任务没有结果，因此可以定义为Runnable。然而，方便的ExecutorService#invokeAll方法只接受Callable对象。因此，我们使用Executors.callable方法为每个Runnable对象创建一个Callable对象。

保护:通过线程发送到System.out的输出不一定按时间顺序出现在控制台上。总是包括时间戳(如Instant.now() )。如果你在乎顺序，研究一下那些时间戳。或者使用与System.out不同的报告机制。

确保正确关闭您的执行者服务。请参阅shutdownAndAwaitTermination方法中Javadoc中的样板代码。我在这里稍微修改了代码。注意，ExecutorService#awaitTermination阻塞直到所有提交的任务完成。

在Java20中现在预览的未来特性中，https://wiki.openjdk.org/display/loom/Main可能会带来使用结构化并发执行一组任务的另一种方式。如果好奇，请参阅相关的JavaJples425& 428。

package work.basil.example.tsk;

import java.time.*;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;

public class App
{
    public static void main ( String[] args )
    {
        App app = new App();
        app.demo();
    }

    private void demo ( )
    {
        ExecutorService executorService = Executors.newFixedThreadPool( 3 );
        List < Future < Object > > futures;
        try
        {
            futures = executorService.invokeAll( this.getTasks() );
        }
        catch ( InterruptedException e )
        {
            throw new RuntimeException( e );
        }
        this.shutdownAndAwaitTermination( executorService );
    }

    private Collection < Callable < Object > > getTasks ( )
    {
        List < String > inputs = List.of( "firstTask" , "secondTask" , "thirdTask" , "foiuthTask" , "fifthTask" );

        List < Callable < Object > > tasks =
                inputs
                        .stream()
                        .map( input -> new StringProcessor( input ) )  // Instantiate a task for each input, a `StringProcessor` implementing `Runnable` interface.
                        .map( Executors :: callable ) // Make `Callable` objects from our `Runnable` objects, to appease the  `ExecutorService#invokeAll` method.
                        .toList();
        return tasks;
    }

    void shutdownAndAwaitTermination ( ExecutorService executorService )
    {
        executorService.shutdown(); // Disable new tasks from being submitted
        try
        {
            // Wait a while for existing tasks to terminate
            if ( ! executorService.awaitTermination( 60 , TimeUnit.SECONDS ) )
            {
                executorService.shutdownNow(); // Cancel currently executing tasks
                // Wait a while for tasks to respond to being cancelled
                if ( ! executorService.awaitTermination( 60 , TimeUnit.SECONDS ) )
                { System.err.println( "Executor service did not terminate. " + Instant.now() ); }
            }
        }
        catch ( InterruptedException ex )
        {
            // (Re-)Cancel if current thread also interrupted
            executorService.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }
}


class StringProcessor implements Runnable
{
    String input;

    public StringProcessor ( final String input )
    {
        this.input = input;
    }

    @Override
    public void run ( )
    {
        System.out.println( "input = " + input + " processed at " + Instant.now() );
    }
}

跑步时：

input = thirdTask processed at 2022-11-03T20:02:22.543733Z
input = firstTask processed at 2022-11-03T20:02:22.543782Z
input = foiuthTask processed at 2022-11-03T20:02:22.555666Z
input = fifthTask processed at 2022-11-03T20:02:22.555675Z
input = secondTask processed at 2022-11-03T20:02:22.543744Z