如何有效管理大量 Java 可运行任务 [关闭]答案

【问题标题】：How to effectively manage a big number of Java runnable tasks [closed]如何有效管理大量 Java 可运行任务 [关闭]
【发布时间】：2021-03-26 08:08:42
【问题描述】：

我有一个包含 100K+ 标识符的数据库表。任务是读取每个 ID 并启动耗时的充实过程，其结果一旦完成，需要持久化到数据库中。

当前基于 Java 的解决方案基于以下流程：

将数据库中的所有标识符加载到内存中；
创建 10 个线程的 ThreadPool；
启动并行处理/任务，这些任务执行丰富并将数据持久保存到数据库中（每个任务 50 个 ID，并行）。正在将所有任务添加到队列中。

Java ThreadPool 和 Executor 框架用于容纳此功能。

未来的预期是标识符的数量将急剧增长，因此为了避免潜在的内存压力 - 我正在考虑开始批量读取 ID 并在现有任务之一完成时创建相关任务（即按需）。

例如，在开始时，在主线程中，从数据库中获取 500 个 ID，创建 10 个任务（每个任务 50 个 ID），将它们传递给工作线程来处理它们，一旦第一个任务完成 - 提取额外的 ID从数据库中创建一个附加任务。然后重复该过程，直到处理完所有 ID。

我的问题是如何通知主线程特定任务/任务已完成，以允许主线程从数据库中提取额外的 ID 并创建额外的任务？这是解决此问题的最佳方法还是在这种情况下其他架构会更好？

【问题讨论】：

您预计工作的哪一部分会出现记忆问题？创建 Runnable/Callable 任务对象的实例？
谢谢，会改的。

标签： java multithreading concurrency threadpool executor

【解决方案1】：

当您将 Runnable/Callable 对象提交给执行器服务时，您会得到一个 Future 对象。您可以跟踪那些Future 对象，并询问它们的状态。每个人都会报告它是否被取消或完成。

全部完成后，向执行器服务提交另一批 Runnable/Callable 任务。

你可以在后台线程上运行这个Future-checker batch-submitter 作为另一个任务，使用ScheduledExecutorService 重复执行。不直接涉及主线程。

除此之外，我建议您检查您的假设。显然，您担心实例化数百万个 Runnable/Callable 对象，以免耗尽内存。但我怀疑每个 Runnable/Callable 对象以及生成的 Future 对象都占用大量内存。我建议您运行模拟来查看并使用监视器或分析工具检查内存使用情况。

这是一些示例代码。首先是我的Callable。

package work.basil.example;

import java.util.concurrent.Callable;

public record Enrichment(Integer id) implements Callable
{
    @Override
    public Boolean call ( ) throws Exception
    {
        System.out.println( this.toString() );
        return Boolean.TRUE; // Report success.
    }
}

还有一些代码可以预定 1000 万个可运行实例，并累积提交到执行器服务时产生的每个 Future 对象。

Instant start = Instant.now();
System.out.println( "INFO - Start running demo at " + start );

int limit = 10_000_000;
List < Future > futures = new ArrayList <>( limit );
ExecutorService executorService = null;
try
{
    executorService = Executors.newFixedThreadPool( 3 );
    for ( int i = 1 ; i <= limit ; i++ )
    {
        Callable < Boolean > callable = new Enrichment( i );
        Future < Boolean > future = executorService.submit( callable );
        futures.add( future );
    }
    System.out.println( "INFO - Submitted %d tasks.".formatted( limit ) );
}
finally
{
    if ( Objects.nonNull( executorService ) ) { executorService.shutdown(); }
}

// Sleep our main thread long enough for background work to finish.
try
{
    System.out.println( "INFO - Sleeping main thread." );
    Thread.sleep( TimeUnit.MINUTES.toMillis( 1 ) );
}
catch ( InterruptedException e )
{
    e.printStackTrace();
}

Instant done = Instant.now();
System.out.println( "INFO - Done running demo at " + done );

请注意，在Enrichment 类的这种特殊情况下，我不需要真正实例化新对象。我们可以简单地在 run 的所有 1000 万次执行中重复使用单个实例。但我想要一个更糟糕的例子——如果你的场景确实需要新的对象，我想看看对内存的大致影响。

在我在 64 位 Intel Mac mini 上使用来自 AdoptOpenJDK 的 Java 15 的试验中，这项工作花费了不到一分钟的时间，并使用 4.5 gigs 处理了 1000 万个任务。

顺便说一句，将来Project Loom 可能会简化您的工作。您将能够简单地安排数百万个“虚拟线程”（纤程）在有限数量的平台/内核线程上运行。 Project Loom 抢先体验Builds are available now。在 YouTube 上查看 Ron Pressler 2020 年末的演讲。

【讨论】：

谢谢您，Basil，您的评论。主要担心的是，否则所有 ID（数百万或记录）都需要一次性加载到内存中，并且需要排队大量可运行的任务同时。我在想，与其等待整个批次（例如 10 个任务）完成，不如分别检查每个未来，以便在第一个完成时立即创建一个新任务来处理新的 ID 集。如果 future 失败（远程服务超时），然后尝试重新运行它。它是否会使实现逻辑过于复杂（或者它根本不是最佳实践）？
@Dmytro 我没有看到使用执行程序服务对数百万个可运行/可调用任务进行排队的问题，而不是在阻塞队列方面。排队任务是执行器服务的核心工作。我不知道数百万范围内的任何特定限制。但我想你可以仔细阅读源代码。并尝试一个实验，正如我在答案中建议的那样。
@Dmytro 关于如何处理失败的任务，我不知道这方面的最佳实践。

【解决方案2】：

@Basil Bourque 的方法非常好，我有另一种方法希望可以帮助到你。

任务进程需要更多时间，因此无需将所有 IDS 从数据库加载到内存。
您必须有一个批量查询方法，每次可以从数据库中获取 500 个 ID。
使用方法调用实现通知主线程。

我为此写了一个非常简单的代码：

public class BatchTaskExecutor {

    // contains method about query from database
    private final TaskRepository taskRepository;

    private final List<List<Task>> needProcessedTasks = new LinkedList<>();

    private final ExecutorService executorService = Executors.newSingleThreadExecutor();

    public BatchTaskExecutor(TaskRepository taskRepository) {
        this.taskRepository = taskRepository;
    }

    public void process() {
        executorService.submit(() -> {
            loadTasks();
            Mono<Boolean> processResult = taskProcess();
            processResult.block();
        });
    }

    // load tasks from database and partition to ten list for parallel process 
    private void loadTasks() {
        List<Task> tasks = taskRepository.batchQuery();

        List<List<Task>> partitionTasks = Lists.partition(tasks, 50);
        needProcessedTasks.clear();
        needProcessedTasks.addAll(partitionTasks);
    }

    // use reactor to parallel process
    private Mono<Boolean> taskProcess() {
        return Flux.range(0, 10)
            .parallel(10)
            .runOn(Schedulers.parallel())
            .flatMap((Function<Integer, Publisher<List<Task>>>) integer -> Mono.just(needProcessedTasks.get(integer)))
            .flatMap((Function<List<Task>, Publisher<Boolean>>) tasks -> Mono.just(batchProcessTask(tasks))).reduce(new BiFunction<Boolean, Boolean, Boolean>() {
            @Override
            public Boolean apply(Boolean aBoolean, Boolean aBoolean2) {
                return aBoolean && aBoolean2;
            }
        }).doOnSuccess(aBoolean -> {
            if (aBoolean) {
                batchSave();
                process();
            } else {
                // todo realize retry method
            }
        });
    }

    // simulate long time need task
    private boolean batchProcessTask(List<Task> tasks) {
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException ignore) {
        }
        for (Task task : tasks) {
            task.setStatus(true);
        }
        return true;
    }

    private void batchSave() {
        System.out.println("begin to save task");
        taskRepository.batchSave(needProcessedTasks.stream()
            .flatMap(new Function<List<Task>, Stream<Task>>() {
                @Override
                public Stream<Task> apply(List<Task> tasks) {
                    return tasks.stream();
                }
            }).collect(Collectors.toList()));

        needProcessedTasks.clear();
    }
}

【讨论】：