java - 使用Java并发API建模动态数据流的技术

Question

编辑：这基本上是一个“如何在Java中正确实现数据流引擎”的问题，我觉得这不能在一个答案中得到充分回答（这就像问“如何正确实现ORM层”并使某人写出Hibernate的详细信息或其他内容），因此请将此问题视为“已关闭”。

有没有一种优雅的方法可以在Java中为动态数据流建模？通过数据流，我的意思是存在各种类型的任务，并且这些任务可以任意“连接”，这样，当一个任务完成时，后续任务将使用完成的任务输出作为输入并行执行，或者当多个任务完成时，它们输出汇总在后续任务中（请参见flow-based programming）。动态的意思是，任务完成时后继任务的类型和数量取决于该完成任务的输出，因此，例如，任务A如果有一定的输出，就可以生成任务B，但是如果任务B具有一定的输出，则可以生成任务C。不同的输出。换句话说，每个任务（或一组任务）负责确定下一个任务是什么。

用于呈现网页的示例数据流：我具有以下任务类型：文件下载器，HTML / CSS呈现器，HTML解析器/ DOM构建器，图像呈现器，JavaScript解析器，JavaScript解释器。


HTML文件的文件下载器任务


HTML解析器/ DOM构建器任务


每个嵌入式文件/链接的文件下载器任务


如果是图像，则为图像渲染器
如果使用外部JavaScript，则JavaScript解析器


JavaScript解释器

否则，只需将其存储在HTML解析器任务的某些var / field中

每个嵌入式脚本的JavaScript解析器


JavaScript解释器

等待上述任务完成，然后等待HTML / CSS渲染器（显然不是最佳方法或完全正确的方法，但这很简单）




我并不是说解决方案需要一个综合的框架（实际上，与JDK API越近越好），而且我绝对不希望这样做，因为重量级的东西是说Spring Web Flow或一些声明性标记或其他DSL。 。

更具体地说，我正在尝试一种使用Callables，Executors，ExecutorCompletionServices以及各种同步器类（例如Semaphore或CountDownLatch）在Java中对此建模的好方法。有几个用例和要求：


不要对任务将在哪些执行程序上运行做任何假设。实际上，为简化起见，只需假设只有一个执行程序。它可以是固定的线程池执行器，因此幼稚的实现会导致死锁（例如，假设一个任务提交了另一个任务，然后阻塞直到该子任务完成，现在想象其中的几个任务用尽了所有线程）。
为简化起见，假设没有在任务之间传输数据（任务输出->后续任务输入）-结束任务和后续任务不必并存，因此后续任务的输入数据将不会被更改之前的任务（因为它已经完成）。
数据流“引擎”应该只能处理几个操作：


一种任务可以排队更多任务的机制
一种机制，直到所有必需的输入任务都完成后，后续任务才排队
主线程（或其他未由执行程序管理的线程）阻塞直到完成流程的机制
主线程（或其他未由执行者管理的线程）阻塞直到某些任务完成的机制

由于数据流是动态的（取决于任务的输入/状态），因此这些机制的激活应在任务代码内进行，例如Callable中的代码本身负责对更多Callable进行排队。
数据流“内部”不应暴露给任务（可调用对象）本身-只有上面列出的操作才对任务可用。
请注意，所有任务的数据类型不一定相同，例如文件下载任务可以接受文件作为输入，但将输出字符串。
如果任务抛出未捕获的异常（表明某些致命错误，要求停止所有数据流处理），则该任务必须传播到尽快启动数据流的线程，并取消所有任务（或像致命错误处理程序之类的幻想者）。
任务应尽快启动。这与先前的要求一起应排除简单的Future轮询+ Thread.sleep（）。
另外，我希望数据流引擎本身在每次任务完成时或自上一个任务完成以来的X时间内没有完成时执行一些操作（例如记录）。类似于：ExecutorCompletionService<T> ecs; while (hasTasks()) { Future<T> future = ecs.poll(1 minute); some_action_like_logging(); if (future != null) { future.get() ... } ... }


有没有简单的方法可以使用Java并发API来完成所有这些工作？还是不管JDK中提供的功能如何，它都会变得很复杂，是否有一个满足要求的轻量级库？我已经有一个适合我的特定用例的局部解决方案（它以某种方式作弊，因为我正在使用两个执行器，所以要知道，它根本与我上面给出的Web浏览器示例无关），但是我希望看到更通用，更优雅的解决方案。

Answer 1

如何定义接口，例如：

interface Task extends Callable {
  boolean isReady();
}

然后，您的“数据流引擎”将只需要管理一组Task对象，即允许将新Task对象排队执行，并允许查询给定任务的状态（因此，上面的接口可能需要扩展以包括id和/或键入）。当任务完成时（当然，并且在引擎启动时），引擎必须仅查询任何未启动的任务以查看它们现在是否准备就绪，如果可以，则将其传递给执行程序。如您所述，任何日志记录等也可以完成。

可能会有帮助的另一件事是使用Guice（ http://code.google.com/p/google-guice/）或类似的轻量级DI框架来帮助正确连接所有对象（例如，确保创建了正确的执行程序类型，并确保需要访问的Tasks可以为数据流引擎提供一个实例（例如，针对其isReady方法或用于排队其他任务），而无需引入复杂的循环关系。

HTH，但如果我错过任何关键方面，请发表评论...
保罗