自定义任务执行器或“我在重新发明轮子吗？”答案

【问题标题】：Custom task Executor or "am I reinventing the wheel?"自定义任务执行器或“我在重新发明轮子吗？”
【发布时间】：2014-08-06 05:56:15
【问题描述】：

我正在尝试创建一个实用程序类，该类将能够处理可运行包并以不同的组合（同步、异步）执行它们。

例如：假设这是一个类似 json 的复合任务表示。 “[]”保持异步任务，“{}”保持同步任务。

[
    task,
    [task, task, task],
    {task, task, [task, task]}
]

这是（将）以方法链接方式实现的：

fromAsyncTasks(
    from(runnable),
    fromSyncedTasks(from(runnable), from(runnable), from(runnable)),
    fromAsyncTasks(from(runnable), from(runnable), fromAsyncTasks(from(runnable), from(runnable)))
).execute();

正如您可能理解的那样，.execute() 是以同步或异步方式递归调用其他任务 execute()。 Task 接口还支持 terminate() 停止（中断任务），因此如果我保留对这些任务的引用，我将能够终止它们。

所以问题 #1 是：是否有任何工具、库至少提供此功能？

和#2：如果我使用流进行并行（异步情况）执行，我该如何终止它们？

tasks.parallelStream().forEach(Runnable::run)

【问题讨论】：

invoke a bunch of task that should complete or invoke a bunch of task that should stop once one completed or build composed tasks, …
@Holger 我确实知道 Executor 接口，但问题是会有很多业务逻辑会动态执行一些任务（同步或异步或组合），如果我们几乎不使用 Executors 代码将是可读的。
澄清一下：你的意思是 async 的 inparallel 和 sync的 sequentially >?

标签： java multithreading concurrency java-8 java-stream

【解决方案1】：

从 Java 8 开始，java.util.concurrent.CompletableFuture 类是标准库的一部分。此类基本上提供了构建所需内容所需的一切。

要直接支持您的用例，您需要某种抽象，即Runnable、同步任务列表或异步任务列表。在下面的代码中，这个抽象是类Task：

interface Task {
    CompletableFuture<Void> execute(
        CompletableFuture<Void> f, Executor e);
}

以下是创建三种不同类型任务的三种工厂方法：

static Task wrap(Runnable runnable) {
    return (f, e) -> f.thenRunAsync(runnable, e);
}
static Task sync(Task... tasks) {
    return (f, e) -> {
        for (Task task : tasks) {
            f = task.execute(f, e);
        }
        return f;
    };
}
static Task async(Task... tasks) {
    return (f, e) -> tasks.length == 0 ? f :
        CompletableFuture.allOf(
        Arrays.stream(tasks)
        .map(t -> t.execute(f, e))
        .toArray(CompletableFuture[]::new));
}

现在，您可以轻松地创建可运行、同步和异步任务的任意嵌套结构，并使用任意 Executor 执行它们。这是一个简短的例子：

public static void main(String... args) {
    Task task = sync(
        wrap(() -> log("1")),
        async(
            wrap(() -> log("A")),
            wrap(() -> log("B")),
            wrap(() -> log("C"))),
        wrap(() -> log("2")),
        wrap(() -> log("3")));
    ExecutorService executor = Executors.newFixedThreadPool(4);
    task.execute(CompletableFuture.completedFuture(null), executor).join();
    executor.shutdown();
}

【讨论】：

谢谢，我不知道 CompletableFuture 和 ComplationStage 接口。
如果我写了这个实现，它需要几个类，而这个实现很漂亮，但是代码对我来说看起来很神奇，虽然我理解它的作用我敢肯定我从来没有那样想，也许是因为我从未使用过函数式编程或任何以闭包为中心的语言。您能推荐任何可以促进 lambda 思维方式的书籍或文章吗？
我用过这个，但我不确定如何从已完成的任务中获取价值？
@Vach：您的原始代码中没有返回值，所以我也省略了它们，因为它使代码更加复杂（特别是关于 async 执行） .尽管如此，CompletableFuture 的 type 参数指定了返回类型，join 方法返回结果。
@Vach：我不知道有哪本书涵盖了 Java 8 特性的上述用法。该代码非常特定于 Java 8。一本关于 lambdas 的一般书籍将无济于事。也许将来会有类似 More Effective Java 的东西。

【解决方案2】：

某些任务在其他任务之后执行一定是有原因的。这些原因应该以某种方式表示 - 作为消息或信号。当它们被明确表示时，您的类似 json 的符号可以转换为 Dataflow diagram。

有关执行数据流图的库，请查看我的df4j2。

【讨论】：