任务

本节将更详细地介绍如何在 Spring Cloud Data Flow 上编排 Spring Cloud Task 应用程序。spring-doc.cadn.net.cn

如果你刚刚开始使用 Spring Cloud Data Flow，建议在深入阅读本节之前，先阅读“本地（Local）”、“Cloud Foundry”或“Kubernetes”的入门指南。spring-doc.cadn.net.cn

25. 简介

任务型应用是短生命周期的，意味着它会主动停止运行，并可按需执行或安排在稍后时间运行。 一个使用场景可能是抓取网页内容并写入数据库。spring-doc.cadn.net.cn

Spring Cloud Task框架基于Spring Boot构建，并为Boot应用程序增加了记录短生命周期应用生命周期事件的能力，例如启动时间、结束时间和退出状态。 TaskExecution文档展示了数据库中存储了哪些信息。 在Spring Cloud Task应用程序中，代码执行的入口点通常是Boot的CommandLineRunner接口的实现，如本示例所示。spring-doc.cadn.net.cn

对于编写短期运行应用程序的 Spring 开发者来说，Spring Batch 项目可能是首先想到的解决方案。 Spring Batch 提供的功能比 Spring Cloud Task 丰富得多，在处理大量数据时推荐使用。 一个典型的使用场景可能是读取多个 CSV 文件，转换每一行数据，并将转换后的每一行写入数据库。 Spring Batch 提供了自己的数据库模式，其中包含关于 Spring Batch 作业执行的更为丰富的信息集合。 Spring Cloud Task 与 Spring Batch 集成，因此如果一个 Spring Cloud Task 应用程序定义了一个 Spring Batch Job，就会在 Spring Cloud Task 和 Spring Batch 的执行表之间建立关联。spring-doc.cadn.net.cn

在本地机器上运行 Data Flow 时，任务（Tasks）会在一个独立的 JVM 中启动。 在 Cloud Foundry 上运行时，任务会使用Cloud Foundry 的任务（Task）功能来启动。在 Kubernetes 上运行时，任务会通过 Pod 或 Job 资源来启动。spring-doc.cadn.net.cn

26. 任务的生命周期

在深入探讨创建任务的细节之前，你应该了解在 Spring Cloud Data Flow 上下文中任务的典型生命周期：spring-doc.cadn.net.cn

  1. 创建一个任务应用程序spring-doc.cadn.net.cn

  2. 注册任务应用spring-doc.cadn.net.cn

  3. 创建任务定义spring-doc.cadn.net.cn

  4. 启动一个任务spring-doc.cadn.net.cn

  5. 查看任务执行情况spring-doc.cadn.net.cn

  6. 销毁任务定义spring-doc.cadn.net.cn

  7. 持续部署spring-doc.cadn.net.cn

26.1. 创建一个任务应用

虽然 Spring Cloud Task 确实提供了一些开箱即用的应用程序（位于 spring-cloud-task-app-starters），但大多数任务应用程序仍需进行定制开发。 要创建一个自定义的任务应用程序：spring-doc.cadn.net.cn

  1. 使用Spring Initializer创建一个新项目，并确保选择以下Starters（starters）：spring-doc.cadn.net.cn

    1. Cloud Task：此依赖项为 spring-cloud-starter-task。spring-doc.cadn.net.cn

    2. JDBC：此依赖项是 spring-jdbc 起步依赖。spring-doc.cadn.net.cn

    3. 选择您的数据库依赖项：输入 Data Flow 当前正在使用的数据库依赖项。例如：H2。spring-doc.cadn.net.cn

  2. 在您的新项目中，创建一个新类作为主类，如下所示：spring-doc.cadn.net.cn

    @EnableTask
@SpringBootApplication
public class MyTask {

    public static void main(String[] args) {
		SpringApplication.run(MyTask.class, args);
	}
}

  3. 使用此类时，您的应用程序中需要一个或多个 CommandLineRunner 或 ApplicationRunner 的实现。您可以自行实现，也可以使用 Spring Boot 提供的实现（例如，有一个用于运行批处理作业的实现）。spring-doc.cadn.net.cn

  4. 使用 Spring Boot 将您的应用程序打包成一个 uber jar，是通过标准的Spring Boot 约定来完成的。 打包后的应用程序可以按照如下所述进行注册和部署。spring-doc.cadn.net.cn

26.1.1. 任务数据库配置

启动任务应用程序时，请确保 Spring Cloud Data Flow 所使用的数据库驱动程序也作为依赖项包含在任务应用程序中。 例如，如果您的 Spring Cloud Data Flow 配置为使用 PostgreSQL，请确保任务应用程序也将 PostgreSQL 作为依赖项。
当你从外部（即从命令行）运行任务，并希望 Spring Cloud Data Flow 在其 UI 中显示 TaskExecutions 时，请确保两者之间共享相同的数据库配置。 默认情况下，Spring Cloud Task 使用本地的 H2 实例，而执行记录会被写入 Spring Cloud Data Flow 所使用的数据库中。

26.2. 注册任务应用

你可以使用 Spring Cloud Data Flow Shell 的 app register 命令将一个 Task 应用注册到应用注册表（App Registry）中。 你必须提供一个唯一名称以及一个可解析到应用制品（artifact）的 URI。对于类型，请指定为 task。 以下列出了三个示例：spring-doc.cadn.net.cn

dataflow:>app register --name task1 --type task --uri maven://com.example:mytask:1.0.2

dataflow:>app register --name task2 --type task --uri file:///Users/example/mytask-1.0.2.jar

dataflow:>app register --name task3 --type task --uri https://example.com/mytask-1.0.2.jar

当提供带有 maven 方案的 URI 时，其格式应符合以下规范：spring-doc.cadn.net.cn

maven://<groupId>:<artifactId>[:<extension>[:<classifier>]]:<version>

如果你想一次性注册多个应用程序，可以将它们存储在一个属性文件中，其中键的格式为 <type>.<name>，值为对应的 URI。 例如，以下内容就是一个有效的属性文件：spring-doc.cadn.net.cn

task.cat=file:///tmp/cat-1.2.1.BUILD-SNAPSHOT.jar
task.hat=file:///tmp/hat-1.2.1.BUILD-SNAPSHOT.jar

然后，您可以使用 app import 命令，并通过 --uri 选项提供属性文件的位置，如下所示：spring-doc.cadn.net.cn

app import --uri file:///tmp/task-apps.properties

例如，如果你想通过一次操作注册 Data Flow 自带的所有任务应用，可以使用以下命令：spring-doc.cadn.net.cn

dataflow:>app import --uri https://dataflow.spring.io/task-maven-latest

你也可以传递 --local 选项（默认值为 TRUE），以指示是否应在 Shell 进程内部解析属性文件的位置。 如果应从 Data Flow Server 进程中解析该位置，请指定 --local false。spring-doc.cadn.net.cn

在使用 app register 或 app import 时，如果所提供的名称和版本的任务应用已经注册，默认情况下不会被覆盖。如果您希望使用不同的 uri 或 uri-metadata 位置来覆盖已存在的任务应用，请包含 --force 选项。spring-doc.cadn.net.cn

在某些情况下，资源会在服务器端进行解析。 在其他情况下，URI 会被传递给运行时容器实例，并在那里进行解析。 有关更多详细信息，请参阅各个 Data Flow Server 的具体文档。

26.3. 创建任务定义

你可以通过提供一个定义名称以及适用于任务执行的属性，从任务应用程序创建一个任务定义。你可以通过 RESTful API 或 Shell 来创建任务定义。要使用 Shell 创建任务定义，请使用 task create 命令来创建任务定义，如下例所示：spring-doc.cadn.net.cn

dataflow:>task create mytask --definition "timestamp --format=\"yyyy\""
Created new task 'mytask'

您可以通过 RESTful API 或 shell 获取当前任务定义的列表。 要使用 shell 获取任务定义列表，请使用 task list 命令。spring-doc.cadn.net.cn

26.3.1. 任务定义名称的最大长度

任务定义名称的最大字符长度取决于平台。spring-doc.cadn.net.cn

请查阅平台文档以了解有关资源命名的具体信息。 Local 平台将任务定义名称存储在一个最大长度为 255 的数据库列中。
表1. 各平台任务定义名称的最大字符长度
Kubernetes 裸 Pod Kubernetes 作业 Cloud Foundry 本地

63spring-doc.cadn.net.cn

52spring-doc.cadn.net.cn

63spring-doc.cadn.net.cn

255spring-doc.cadn.net.cn

26.3.2. 自动创建任务定义

从 2.3.0 版本开始，您可以通过将 spring.cloud.dataflow.task.autocreate-task-definitions 设置为 true 来配置 Data Flow 服务器自动创建任务定义。 这并非默认行为，而是作为一种便利功能提供。 启用此属性后，任务启动请求可以将已注册的任务应用程序名称直接指定为任务名称。 如果该任务应用程序已注册，服务器将创建一个仅包含应用程序名称的基本任务定义（这是必需的）。这样就省去了类似以下的手动操作步骤：spring-doc.cadn.net.cn

dataflow:>task create mytask --definition "mytask"

您仍然可以为每次任务启动请求指定命令行参数和部署属性。spring-doc.cadn.net.cn

26.4. 启动一个任务

可以通过 RESTful API 或 Shell 启动一个临时任务。 要通过 Shell 启动临时任务，请使用 task launch 命令，如下例所示：spring-doc.cadn.net.cn

dataflow:>task launch mytask
Launched task 'mytask'

启动任务时，您可以设置在启动任务应用程序时需要作为命令行参数传递的任何属性，如下所示：spring-doc.cadn.net.cn

dataflow:>task launch mytask --arguments "--server.port=8080 --custom=value"
参数需要以空格分隔的值形式传递。

你可以使用 TaskLauncher 选项传入专用于 --properties 本身的附加属性。 该选项的格式是一个以逗号分隔的属性字符串，每个属性都带有前缀 app.<task definition name>.<property>。 这些属性会作为应用程序属性传递给 TaskLauncher。 具体实现可自行决定如何将这些属性传递给实际的任务应用程序。 如果属性前缀是 deployer 而非 app，则该属性会作为部署属性传递给 TaskLauncher，其具体含义可能因 TaskLauncher 的实现而异。spring-doc.cadn.net.cn

dataflow:>task launch mytask --properties "deployer.timestamp.custom1=value1,app.timestamp.custom2=value2"

26.4.1. 应用程序属性

每个应用程序都通过属性来定制其行为。例如，timestamp 任务的 format 设置可以指定一个不同于默认值的输出格式。spring-doc.cadn.net.cn

dataflow:> task create --definition "timestamp --format=\"yyyy\"" --name printTimeStamp

此 timestamp 属性实际上与 timestamp 应用程序中指定的 timestamp.format 属性相同。 Data Flow 添加了使用简写形式 format 来替代 timestamp.format 的能力。 您也可以指定完整形式，如下例所示：spring-doc.cadn.net.cn

dataflow:> task create --definition "timestamp --timestamp.format=\"yyyy\"" --name printTimeStamp

这一简写行为在流应用属性一节中有更详细的讨论。 如果你已经注册了应用属性元数据，那么在 Shell 中输入 -- 后可以使用 Tab 键自动补全，以获取候选属性名称列表。spring-doc.cadn.net.cn

Shell 为应用程序属性提供制表符自动补全功能。app info --name <appName> --type <appType> Shell 命令为所有支持的属性提供额外的文档说明。支持的任务 <appType> 为 task。spring-doc.cadn.net.cn

在 Kubernetes 上重启 Spring Batch 作业时，必须使用 shell 或 boot 作为入口点。
在 Kubernetes 上包含敏感信息的应用程序属性

在启动任务应用程序时，如果某些属性可能包含敏感信息，请使用 shell 或 boot 作为 entryPointStyle。这是因为默认的 exec 方式会将所有属性转换为命令行参数，因此在某些环境中可能不安全。spring-doc.cadn.net.cn

26.4.2. 常见应用程序属性

除了通过 DSL 进行配置外，Spring Cloud Data Flow 还提供了一种机制，用于设置由它启动的所有任务应用程序所共用的属性。 您可以在启动服务器时添加以 spring.cloud.dataflow.applicationProperties.task 为前缀的属性来实现这一点。 随后，服务器会将这些属性（去掉前缀后）传递给它所启动的实例。spring-doc.cadn.net.cn

例如，您可以通过使用以下选项启动 Data Flow 服务器，来配置所有已启动的应用程序使用 prop1 和 prop2 属性：spring-doc.cadn.net.cn

--spring.cloud.dataflow.applicationProperties.task.prop1=value1
--spring.cloud.dataflow.applicationProperties.task.prop2=value2

这将导致 prop1=value1 和 prop2=value2 属性被传递给所有启动的应用程序。spring-doc.cadn.net.cn

通过此机制配置的属性优先级低于任务部署属性。 如果在任务启动时指定了相同键的属性（例如，app.trigger.prop2 会覆盖通用属性），则这些属性将被覆盖。

26.5. 限制并发任务启动的数量

Spring Cloud Data Flow 允许用户为每个已配置的云平台限制并发运行的任务最大数量，以防止 IaaS 或硬件资源饱和。 默认情况下，所有受支持平台的此限制均设置为 20。如果某个平台实例上并发运行的任务数大于或等于该限制，则下一个任务启动请求将失败，并通过 RESTful API、Shell 或 UI 返回错误消息。 您可以通过设置相应的部署器属性来为平台实例配置此限制，即 spring.cloud.dataflow.task.platform.<platform-type>.accounts[<account-name>].maximumConcurrentTasks，其中 <account-name> 是已配置的平台账户名称（如果未显式配置任何账户，则为 default）。 <platform-type> 指的是当前支持的部署器之一：local 或 kubernetes。对于 cloudfoundry，该属性为 spring.cloud.dataflow.task.platform.<platform-type>.accounts[<account-name>].deployment.maximumConcurrentTasks。（区别在于 deployment 已添加到路径中）。spring-doc.cadn.net.cn

每个受支持平台的 TaskLauncher 实现会（如果可能）通过查询底层平台的运行时状态来确定当前正在运行的任务数量。识别一个 task 的方法因平台而异。例如，在本地主机上启动任务使用的是 LocalTaskLauncher。LocalTaskLauncher 每次启动请求都会运行一个过程，并在内存中跟踪这些过程。在这种情况下，我们不会查询底层操作系统，因为通过这种方式识别任务是不切实际的。对于 Cloud Foundry 而言，任务是其部署模型所支持的核心概念。所有任务的状态）可直接通过 API 获取。这表示，账户组织和空间中的每个正在运行的任务容器都会计入运行执行次数，无论它是通过使用Spring Cloud Data Flow启动的，还是直接调用了CloudFoundryTaskLauncher。对于Kubernetes，通过KubernetesTaskLauncher启动一个任务如果成功，将创建一个运行中的pod，我们期望它最终完成或失败。在此环境中，通常没有简单的方法来识别与某个任务对应的 Pod。出于这个原因，我们只计算由KubernetesTaskLauncher启动的pod。由于任务Starters在pod的元数据中提供了task-name标签，我们通过该标签的存在来过滤所有正在运行的pod。spring-doc.cadn.net.cn

26.6. 审查任务执行

任务启动后，其状态将存储在关系型数据库中。该状态包括：spring-doc.cadn.net.cn

您可以通过 RESTful API 或命令行 shell 查看任务执行的状态。 要通过命令行 shell 显示最新的任务执行情况，请使用 task execution list 命令。spring-doc.cadn.net.cn

若要仅获取某个任务定义的任务执行列表，请添加 --name 及该任务定义的名称——例如，task execution list --name foo。若要检索某次任务执行的完整详细信息，请使用 task execution status 命令并指定该任务执行的 ID，例如 task execution status --id 549。spring-doc.cadn.net.cn

26.7. 销毁任务定义

销毁任务定义会将该定义从定义仓库中移除。 这可以通过 RESTful API 或命令行 shell 完成。 要通过命令行 shell 销毁任务，请使用 task destroy 命令，如下例所示：spring-doc.cadn.net.cn

dataflow:>task destroy mytask
Destroyed task 'mytask'

task destroy 命令还提供了一个 cleanup 选项，用于清理即将被销毁任务的任务执行记录，如下例所示：spring-doc.cadn.net.cn

dataflow:>task destroy mytask --cleanup
Destroyed task 'mytask'

默认情况下，cleanup 选项设置为 false（即，默认情况下，任务销毁时不会清理任务执行记录）。spring-doc.cadn.net.cn

要通过 shell 销毁所有任务，请使用 task all destroy 命令，如下例所示：spring-doc.cadn.net.cn

dataflow:>task all destroy
Really destroy all tasks? [y, n]: y
All tasks destroyed

如有需要，您可以使用强制开关：spring-doc.cadn.net.cn

dataflow:>task all destroy --force
All tasks destroyed

该任务定义先前启动的任务的执行信息仍保留在任务仓库中。spring-doc.cadn.net.cn

这不会停止此定义的任何当前正在运行的任务。相反，它会从数据库中移除该任务定义。

task destroy <task-name> 仅删除定义，而不会删除部署在 Cloud Foundry 上的任务。 删除该任务的唯一方法是通过 CLI 分两步操作：spring-doc.cadn.net.cn

+ . 使用 cf apps 命令获取应用列表。 . 确定要删除的任务应用，然后运行 cf delete <task-name> 命令。spring-doc.cadn.net.cn

26.8. 验证任务

有时，任务定义中包含的应用程序在其注册信息中存在无效的 URI。 这可能是由于在应用程序注册时输入了无效的 URI，或者是因为该应用程序已从原本应从中获取的仓库中被移除。 要验证任务中包含的所有应用程序是否均可解析，请使用 validate 命令，如下所示：spring-doc.cadn.net.cn

dataflow:>task validate time-stamp
╔══════════╤═══════════════╗
║Task Name │Task Definition║
╠══════════╪═══════════════╣
║time-stamp│timestamp      ║
╚══════════╧═══════════════╝


time-stamp is a valid task.
╔═══════════════╤═════════════════╗
║   App Name    │Validation Status║
╠═══════════════╪═════════════════╣
║task:timestamp │valid            ║
╚═══════════════╧═════════════════╝

在前面的示例中，用户已验证其时间戳任务。task:timestamp 应用程序是有效的。 现在我们可以看看，如果流定义中包含一个已注册但 URI 无效的应用程序时会发生什么情况：spring-doc.cadn.net.cn

dataflow:>task validate bad-timestamp
╔═════════════╤═══════════════╗
║  Task Name  │Task Definition║
╠═════════════╪═══════════════╣
║bad-timestamp│badtimestamp   ║
╚═════════════╧═══════════════╝


bad-timestamp is an invalid task.
╔══════════════════╤═════════════════╗
║     App Name     │Validation Status║
╠══════════════════╪═════════════════╣
║task:badtimestamp │invalid          ║
╚══════════════════╧═════════════════╝

在这种情况下，Spring Cloud Data Flow 声明该任务无效，因为 task:badtimestamp 具有无效的 URI。spring-doc.cadn.net.cn

26.9. 停止任务执行

在某些情况下，由于平台本身或应用程序业务逻辑的问题，正在平台上运行的任务可能无法停止。 针对此类情况，Spring Cloud Data Flow 提供了向平台发送请求以终止任务的功能。 为此，请为指定的一组任务执行提交一个 task execution stop 请求，如下所示：spring-doc.cadn.net.cn

task execution stop --ids 5

Request to stop the task execution with id(s): 5 has been submitted

通过上述命令，停止执行 id=5 的触发请求被提交给底层的部署器实现。因此，该操作会停止对应的任务。当我们查看该任务执行的结果时，可以看到该任务已成功完成，并返回了退出码 0：spring-doc.cadn.net.cn

dataflow:>task execution list
╔══════════╤══╤════════════════════════════╤════════════════════════════╤═════════╗
║Task Name │ID│         Start Time         │          End Time          │Exit Code║
╠══════════╪══╪════════════════════════════╪════════════════════════════╪═════════╣
║batch-demo│5 │Mon Jul 15 13:58:41 EDT 2019│Mon Jul 15 13:58:55 EDT 2019│0        ║
║timestamp │1 │Mon Jul 15 09:26:41 EDT 2019│Mon Jul 15 09:26:41 EDT 2019│0        ║
╚══════════╧══╧════════════════════════════╧════════════════════════════╧═════════╝

如果您对一个具有关联子任务执行的作业执行（例如组合任务）提交停止请求，系统将为每个子任务执行分别发送停止请求。spring-doc.cadn.net.cn

当停止一个正在运行 Spring Batch 作业的任务执行时，该作业的批处理状态将保持为 STARTED。 当请求停止时，所有支持的平台都会向任务应用程序发送 SIG-INT 信号。这使得 Spring Cloud Task 能够捕获应用程序的状态。然而，Spring Batch 并不会处理 SIG-INT 信号，因此作业虽然停止了，但其状态仍保持为 STARTED。
在启动远程分区的 Spring Batch 任务应用程序时，Spring Cloud Data Flow 支持直接停止工作器分区任务，该功能同时适用于 Cloud Foundry 和 Kubernetes 平台。本地平台不支持停止工作器分区任务。

26.9.1. 停止在 Spring Cloud Data Flow 外部启动的任务执行

你可能希望停止一个在 Spring Cloud Data Flow 之外启动的任务。一个典型的例子是由远程批处理分区应用所启动的工作器（worker）应用。 在这种情况下，远程批处理分区应用会为每个工作器应用存储 external-execution-id。然而，并不会存储任何平台信息。 因此，当 Spring Cloud Data Flow 需要停止一个远程批处理分区应用及其工作器应用时，你需要指定平台名称，如下所示：spring-doc.cadn.net.cn

dataflow:>task execution stop --ids 1 --platform myplatform
Request to stop the task execution with id(s): 1 for platform myplatform has been submitted

27. 订阅任务和批处理事件

当任务启动时，您还可以监听各种任务和批处理事件。 如果启用了任务以生成任务或批处理事件（需添加额外的依赖 spring-cloud-task-stream，以及在使用 Kafka 作为绑定器时，还需添加 spring-cloud-stream-binder-kafka），这些事件将在任务生命周期中发布。 默认情况下，这些已发布事件在消息代理（如 Rabbit、Kafka 等）上的目标名称即为事件本身的名称（例如：task-events、job-execution-events 等）。spring-doc.cadn.net.cn

dataflow:>task create myTask --definition "myBatchJob"
dataflow:>stream create task-event-subscriber1 --definition ":task-events > log" --deploy
dataflow:>task launch myTask

您可以通过在启动任务时指定显式名称来控制这些事件的目标名称，如下所示：spring-doc.cadn.net.cn

dataflow:>stream create task-event-subscriber2 --definition ":myTaskEvents > log" --deploy
dataflow:>task launch myTask --properties "app.myBatchJob.spring.cloud.stream.bindings.task-events.destination=myTaskEvents"

下表列出了代理（broker）上默认的任务和批处理事件及其目标名称：spring-doc.cadn.net.cn

表2. 任务和批处理事件目标

事件spring-doc.cadn.net.cn

目的地spring-doc.cadn.net.cn

任务事件spring-doc.cadn.net.cn

task-eventsspring-doc.cadn.net.cn

作业执行事件spring-doc.cadn.net.cn

job-execution-eventsspring-doc.cadn.net.cn

步骤执行事件spring-doc.cadn.net.cn

step-execution-eventsspring-doc.cadn.net.cn

项目读取事件spring-doc.cadn.net.cn

item-read-eventsspring-doc.cadn.net.cn

项目处理事件spring-doc.cadn.net.cn

item-process-eventsspring-doc.cadn.net.cn

项目写入事件spring-doc.cadn.net.cn

item-write-eventsspring-doc.cadn.net.cn

跳过事件spring-doc.cadn.net.cn

skip-eventsspring-doc.cadn.net.cn

28. 组合任务

Spring Cloud Data Flow 允许您创建一个有向图，其中图中的每个节点都是一个任务应用。 这是通过使用组合任务的 DSL 实现的。 您可以通过 RESTful API、Spring Cloud Data Flow Shell 或 Spring Cloud Data Flow UI 来创建组合任务。spring-doc.cadn.net.cn

28.1. 组合任务运行器

组合任务通过一个名为组合任务运行器（Composed Task Runner）的任务应用程序来执行。Spring Cloud Data Flow 服务器在启动组合任务时会自动部署该组合任务运行器。spring-doc.cadn.net.cn

28.1.1. 配置组合任务运行器

组合任务运行器（composed task runner）应用程序具有一个 dataflow-server-uri 属性，用于验证和启动子任务。 该属性默认值为 localhost:9393。如果您运行的是分布式 Spring Cloud Data Flow 服务器（例如将服务器部署在 Cloud Foundry 或 Kubernetes 上时），则需要提供可用于访问该服务器的 URI。 您可以在启动组合任务时，通过为组合任务运行器应用程序设置 dataflow-server-uri 属性来提供该 URI；或者在启动 Spring Cloud Data Flow 服务器时，通过设置其 spring.cloud.dataflow.server.uri 属性来提供。 在后一种情况下，当启动组合任务时，组合任务运行器应用程序的 dataflow-server-uri 属性会自动被设置。spring-doc.cadn.net.cn

配置选项

ComposedTaskRunner 任务具有以下选项：spring-doc.cadn.net.cn

  • composed-task-arguments 用于每个任务的命令行参数。（字符串，默认值：）。spring-doc.cadn.net.cn

  • increment-instance-enabled 允许在不更改参数的情况下再次运行单个 ComposedTaskRunner 实例，方法是基于上一次执行的 run.id 添加一个递增的编号作业参数。（布尔值，默认值：true）。 ComposedTaskRunner 是使用 Spring Batch 构建的。因此，在成功执行后，批处理作业被视为已完成。 要多次启动相同的 ComposedTaskRunner 定义，必须将 increment-instance-enabled 或 uuid-instance-enabled 属性设置为 true，或者为每次启动更改该定义的参数。 使用此选项时，必须将其应用于目标应用程序的所有任务启动，包括首次启动。spring-doc.cadn.net.cn

  • uuid-instance-enabled 允许通过向 ctr.id 作业参数添加 UUID，在不更改参数的情况下重新运行单个 ComposedTaskRunner 实例。（布尔值，默认：false）。 ComposedTaskRunner 基于 Spring Batch 构建。因此，一旦执行成功，批处理作业即被视为完成。 若要多次启动相同的 ComposedTaskRunner 定义，必须将 increment-instance-enabled 或 uuid-instance-enabled 属性设置为 true，或者在每次启动时更改该定义的参数。 使用此选项时，必须将其应用于目标应用程序的所有任务启动，包括首次启动。将此选项设置为 true 将覆盖 increment-instance-id 的值。 当同时运行同一组合任务定义的多个实例时，请将此选项设置为 true。spring-doc.cadn.net.cn

  • interval-time-between-checks ComposedTaskRunner 在检查数据库以确认任务是否完成时，两次检查之间的等待时间（以毫秒为单位）。（整数，默认值：10000）。 ComposedTaskRunner 使用数据存储来确定每个子任务的状态。此间隔指示 ComposedTaskRunner 应该多久检查一次其子任务的状态。spring-doc.cadn.net.cn

  • transaction-isolation-level 为组合任务运行器（Composed Task Runner）设置事务隔离级别。 可用的事务隔离级别列表请参见此处。 默认值为ISOLATION_REPEATABLE_READ。spring-doc.cadn.net.cn

  • max-wait-time 单个步骤在组合任务（Composed task）执行失败前允许运行的最长时间（以毫秒为单位）（Integer 类型，默认值：0）。 该参数决定每个子任务在组合任务运行器（CTR）因失败而终止之前允许运行的最长时间。默认值 0 表示无超时限制。spring-doc.cadn.net.cn

  • split-thread-allow-core-thread-timeout 指定是否允许拆分的核心线程超时。（布尔值，默认值：false） 设置策略，用于控制核心线程在保持存活时间内如果没有任务到达，是否可以超时并终止；当新任务到达时，如有需要会重新创建线程。spring-doc.cadn.net.cn

  • split-thread-core-pool-size Split 的核心线程池大小。（整数类型，默认为 1） 每个包含在 split 中的子任务都需要一个线程来执行。因此，例如，像 <AAA || BBB || CCC> && <DDD || EEE> 这样的定义将需要 split-thread-core-pool-size 个 3。 这是因为最大的 split 包含三个子任务。数量为 2 意味着 AAA 和 BBB 将并行运行，但 CCC 将等待直到 AAA 或 BBB 完成才能运行。 随后，DDD 和 EEE 将并行运行。spring-doc.cadn.net.cn

  • split-thread-keep-alive-seconds Split 线程的保活时间（秒）。（整数，默认值：60） 如果当前线程池中的线程数量超过 corePoolSize，则超出的线程在空闲时间超过 keepAliveTime 后将被终止。spring-doc.cadn.net.cn

  • split-thread-max-pool-size Split 的最大线程池大小。（整数，默认值：Integer.MAX_VALUE）。 设置线程池允许的最大线程数。spring-doc.cadn.net.cn

  • split-thread-queue-capacity Split 的 BlockingQueue 容量。（Integer 类型，默认值：Integer.MAX_VALUE）spring-doc.cadn.net.cn

    • 如果正在运行的线程数少于 corePoolSize，Executor 总是优先创建新线程，而不是将任务加入队列。spring-doc.cadn.net.cn

    • 如果正在运行的线程数达到或超过 corePoolSize，Executor 总是优先将请求放入队列，而不是创建新线程。spring-doc.cadn.net.cn

    • 如果请求无法加入队列，则会创建一个新线程，除非这样做会超过 maximumPoolSize。在这种情况下，该任务将被拒绝。spring-doc.cadn.net.cn

  • split-thread-wait-for-tasks-to-complete-on-shutdown 在关闭时是否等待已调度的任务完成，不中断正在运行的任务，并执行队列中的所有任务。（布尔值，默认值：false）spring-doc.cadn.net.cn

  • dataflow-server-uri 用于接收任务启动请求的 Data Flow 服务器的 URI。（字符串，默认值：localhost:9393）spring-doc.cadn.net.cn

  • dataflow-server-username 用于接收任务启动请求的 Data Flow 服务器的可选用户名。 用于通过基本认证（Basic Authentication）访问 Data Flow 服务器。如果设置了 dataflow-server-access-token，则不会使用此参数。spring-doc.cadn.net.cn

  • dataflow-server-password 用于接收任务启动请求的 Data Flow 服务器的可选密码。 使用 Basic Authentication（基本认证）访问 Data Flow 服务器时使用。如果设置了 dataflow-server-access-token，则不会使用此参数。spring-doc.cadn.net.cn

  • dataflow-server-access-token 此属性用于设置一个可选的 OAuth2 访问Tokens。 通常，该值会自动使用当前已登录用户的Tokens（如果可用）进行设置。 然而，在某些特殊使用场景下，也可以显式地设置该值。spring-doc.cadn.net.cn

存在一个特殊的布尔属性 dataflow-server-use-user-access-token，用于在您希望使用当前登录用户的访问Tokens并将其传递给组合任务运行器（Composed Task Runner）时。该属性由 Spring Cloud Data Flow 使用，当其值设为 true 时，会自动填充 dataflow-server-access-token 属性。使用 dataflow-server-use-user-access-token 时，必须在每次任务执行时都传递该属性。 在某些情况下，可能更倾向于默认在每次启动组合任务时都传递用户的 dataflow-server-access-token。 此时，请将 Spring Cloud Data Flow 的 spring.cloud.dataflow.task.useUserAccessToken 属性设置为 true。spring-doc.cadn.net.cn

要为 Composed Task Runner 设置一个属性，您需要在属性前加上前缀 app.composed-task-runner.。 例如，要设置 dataflow-server-uri 属性，该属性应写为 app.composed-task-runner.dataflow-server-uri。spring-doc.cadn.net.cn

28.2. 组合任务的生命周期

组合任务的生命周期包含三个部分：spring-doc.cadn.net.cn

28.2.1. 创建组合任务

组合任务的 DSL 在通过 task create 命令创建任务定义时使用，如下例所示：spring-doc.cadn.net.cn

dataflow:> app register --name timestamp --type task --uri maven://org.springframework.cloud.task.app:timestamp-task:
dataflow:> app register --name mytaskapp --type task --uri file:///home/tasks/mytask.jar
dataflow:> task create my-composed-task --definition "mytaskapp && timestamp"
dataflow:> task launch my-composed-task

在前面的示例中，我们假设将要用于组合任务的应用程序尚未注册。 因此，在前两个步骤中，我们注册了两个任务应用程序。 然后，我们使用 task create 命令创建组合任务定义。 前面示例中的组合任务 DSL 在启动时会先运行 mytaskapp，然后再运行 timestamp 应用程序。spring-doc.cadn.net.cn

但在启动 my-composed-task 定义之前，我们可以先查看 Spring Cloud Data Flow 为我们生成了什么内容。 这可以通过使用 task list 命令来实现，如下例所示（包括其输出）：spring-doc.cadn.net.cn

dataflow:>task list
╔══════════════════════════╤══════════════════════╤═══════════╗
║        Task Name         │   Task Definition    │Task Status║
╠══════════════════════════╪══════════════════════╪═══════════╣
║my-composed-task          │mytaskapp && timestamp│unknown    ║
║my-composed-task-mytaskapp│mytaskapp             │unknown    ║
║my-composed-task-timestamp│timestamp             │unknown    ║
╚══════════════════════════╧══════════════════════╧═══════════╝

在该示例中，Spring Cloud Data Flow 创建了三个任务定义：分别为组成组合任务的每个应用程序（my-composed-task-mytaskapp 和 my-composed-task-timestamp）以及组合任务本身（my-composed-task）创建了一个定义。 我们还看到，每个子任务生成的名称均由组合任务的名称和应用程序的名称组成，中间用连字符 - 分隔（例如 my-composed-task - mytaskapp）。spring-doc.cadn.net.cn

任务应用程序参数

组成组合任务定义的任务应用程序也可以包含参数，如下例所示：spring-doc.cadn.net.cn

dataflow:> task create my-composed-task --definition "mytaskapp --displayMessage=hello && timestamp --format=YYYY"

28.2.2. 启动一个组合任务

启动组合任务的方式与启动独立任务相同，如下所示：spring-doc.cadn.net.cn

task launch my-composed-task

一旦任务启动，并假设所有任务都成功完成，当你运行 task execution list 时，就可以看到三次任务执行记录，如下例所示：spring-doc.cadn.net.cn

dataflow:>task execution list
╔══════════════════════════╤═══╤════════════════════════════╤════════════════════════════╤═════════╗
║        Task Name         │ID │         Start Time         │          End Time          │Exit Code║
╠══════════════════════════╪═══╪════════════════════════════╪════════════════════════════╪═════════╣
║my-composed-task-timestamp│713│Wed Apr 12 16:43:07 EDT 2017│Wed Apr 12 16:43:07 EDT 2017│0        ║
║my-composed-task-mytaskapp│712│Wed Apr 12 16:42:57 EDT 2017│Wed Apr 12 16:42:57 EDT 2017│0        ║
║my-composed-task          │711│Wed Apr 12 16:42:55 EDT 2017│Wed Apr 12 16:43:15 EDT 2017│0        ║
╚══════════════════════════╧═══╧════════════════════════════╧════════════════════════════╧═════════╝

在前面的示例中，我们可以看到 my-compose-task 已启动，并且其他任务也按顺序依次启动。 每个任务均成功运行，Exit Code（0）均为 3。spring-doc.cadn.net.cn

将属性传递给子任务

要在任务启动时为组合任务图中的子任务设置属性，请使用以下格式：app.<composed task definition name>.<child task app name>.<property>。 以下清单展示了一个组合任务定义的示例：spring-doc.cadn.net.cn

dataflow:> task create my-composed-task --definition "mytaskapp  && mytimestamp"

要让 mytaskapp 显示 'HELLO'，并将组合任务定义中的 mytimestamp 时间戳格式设置为 YYYY，请使用以下任务启动格式：spring-doc.cadn.net.cn

task launch my-composed-task --properties "app.my-composed-task.mytaskapp.displayMessage=HELLO,app.my-composed-task.mytimestamp.timestamp.format=YYYY"

与应用程序属性类似，你也可以使用以下格式为子任务设置 deployer 属性：deployer.<composed task definition name>.<child task app name>.<deployer-property>：spring-doc.cadn.net.cn

task launch my-composed-task --properties "deployer.my-composed-task.mytaskapp.memory=2048m,app.my-composed-task.mytimestamp.timestamp.format=HH:mm:ss"
Launched task 'a1'
向组合任务运行器传递参数

你可以使用 --arguments 选项为组合任务运行器传递命令行参数：spring-doc.cadn.net.cn

dataflow:>task create my-composed-task --definition "<aaa: timestamp || bbb: timestamp>"
Created new task 'my-composed-task'

dataflow:>task launch my-composed-task --arguments "--increment-instance-enabled=true --max-wait-time=50000 --split-thread-core-pool-size=4" --properties "app.my-composed-task.bbb.timestamp.format=dd/MM/yyyy HH:mm:ss"
Launched task 'my-composed-task'
退出状态

以下列表展示了在组合任务中，每个步骤（任务）执行完成后其退出状态是如何设置的：spring-doc.cadn.net.cn

  • 如果 TaskExecution 包含一个 ExitMessage，则该消息将被用作 ExitStatus。spring-doc.cadn.net.cn

  • 如果没有提供 ExitMessage，并且 ExitCode 设置为零，则该步骤的 ExitStatus 为 COMPLETED。spring-doc.cadn.net.cn

  • 如果没有 ExitMessage，并且 ExitCode 被设置为任意非零数值，则该步骤的 ExitStatus 为 FAILED。spring-doc.cadn.net.cn

28.2.3. 销毁一个组合任务

用于销毁独立任务的命令与用于销毁组合任务的命令相同。 唯一的区别在于，销毁组合任务时还会同时销毁与其关联的子任务。 以下示例展示了使用 destroy 命令前后的任务列表：spring-doc.cadn.net.cn

dataflow:>task list
╔══════════════════════════╤══════════════════════╤═══════════╗
║        Task Name         │   Task Definition    │Task Status║
╠══════════════════════════╪══════════════════════╪═══════════╣
║my-composed-task          │mytaskapp && timestamp│COMPLETED  ║
║my-composed-task-mytaskapp│mytaskapp             │COMPLETED  ║
║my-composed-task-timestamp│timestamp             │COMPLETED  ║
╚══════════════════════════╧══════════════════════╧═══════════╝
...
dataflow:>task destroy my-composed-task
dataflow:>task list
╔═════════╤═══════════════╤═══════════╗
║Task Name│Task Definition│Task Status║
╚═════════╧═══════════════╧═══════════╝

28.2.4. 停止一个组合任务

在需要停止组合任务执行的情况下，您可以通过以下方式实现：spring-doc.cadn.net.cn

要通过仪表板停止一个组合任务，请选择作业选项卡，然后点击您想要停止的作业执行旁边的 *Stop()* 按钮。spring-doc.cadn.net.cn

当当前正在运行的子任务完成时，组合任务运行将停止。 在组合任务停止时正在运行的子任务所对应的步骤，以及组合任务的作业执行（job execution），都会被标记为STOPPED。spring-doc.cadn.net.cn

28.2.5. 重启组合任务

当组合任务在执行过程中失败且其状态为FAILED时，可以重新启动该任务。 您可以通过以下方式实现：spring-doc.cadn.net.cn

要通过命令行重新启动一个组合任务，请使用相同的参数启动该任务。 要通过仪表板重新启动一个组合任务，请选择作业选项卡，然后点击您想要重新启动的作业执行旁边的重新启动按钮。spring-doc.cadn.net.cn

重启一个已被停止的组合任务作业（通过 Spring Cloud Data Flow 仪表板或 RESTful API 停止），会重新启动处于 STOPPED 状态的子任务，然后按照指定顺序启动其余（尚未启动的）子任务。

29. 组合任务 DSL

组合任务可以通过三种方式运行：spring-doc.cadn.net.cn

29.1. 条件执行

条件执行通过使用双与符号（&&）来表示。 这使得序列中的每个任务仅在前一个任务成功完成时才会启动，如下例所示：spring-doc.cadn.net.cn

task create my-composed-task --definition "task1 && task2"

当启动名为 my-composed-task 的组合任务时，它会启动名为 task1 的任务；如果 task1 成功完成，则会启动名为 task2 的任务。 如果 task1 失败，则不会启动 task2。spring-doc.cadn.net.cn

您也可以使用 Spring Cloud Data Flow 仪表板来创建条件执行流程：通过设计器拖放所需的应用程序，并将它们连接起来以构建有向图，如下图所示：spring-doc.cadn.net.cn

Composed Task Conditional Execution
图2. 条件执行

上图是使用 Spring Cloud Data Flow 仪表板创建有向图时的屏幕截图。 您可以看到，图中的四个组件构成了一个条件执行：spring-doc.cadn.net.cn

您可以通过点击“定义”选项卡中组合任务定义旁边的详情按钮，查看有向图的示意图。

29.2. 过渡执行

该 DSL 支持对有向图执行过程中所采取的转换进行细粒度控制。 转换通过提供一个基于前一个任务退出状态的相等性条件来指定。 任务转换由以下符号表示：->。spring-doc.cadn.net.cn

29.2.1. 基本转换

一个基本的过渡效果如下所示：spring-doc.cadn.net.cn

task create my-transition-composed-task --definition "foo 'FAILED' -> bar 'COMPLETED' -> baz"

在前面的示例中，foo 将会启动；如果其退出状态为 FAILED，则 bar 任务将会启动。 如果 foo 的退出状态为 COMPLETED，则 baz 将会启动。 cat 返回的所有其他状态都不会产生影响，任务将正常结束。spring-doc.cadn.net.cn

使用 Spring Cloud Data Flow 仪表板创建相同的“基本转换”将类似于下图所示：spring-doc.cadn.net.cn

Composed Task Basic Transition
图3. 基本转换

上图是 Spring Cloud Data Flow 仪表板中正在创建的有向图的屏幕截图。 请注意，这里有两种不同类型的连接器：spring-doc.cadn.net.cn

  • 虚线：表示从应用程序到其中一个可能的目标应用程序的转换。spring-doc.cadn.net.cn

  • 实线：用于连接在条件执行中的应用程序，或连接应用程序与控制节点（开始或结束）。spring-doc.cadn.net.cn

要创建一个过渡连接器：spring-doc.cadn.net.cn

  1. 创建转换时，请使用连接器将应用程序链接到每个可能的目标。spring-doc.cadn.net.cn

  2. 完成后，依次进入每个连接并点击以选中它。spring-doc.cadn.net.cn

  3. 显示一个螺栓图标。spring-doc.cadn.net.cn

  4. 点击那个图标。spring-doc.cadn.net.cn

  5. 输入该连接器所需的退出状态。spring-doc.cadn.net.cn

  6. 该连接器的实线将变为虚线。spring-doc.cadn.net.cn

29.2.2. 使用通配符的转换

DSL 支持在转换中使用通配符，如下例所示：spring-doc.cadn.net.cn

task create my-transition-composed-task --definition "foo 'FAILED' -> bar '*' -> baz"

在前面的示例中，foo 将会启动；如果其退出状态为 FAILED，则 bar 任务将会启动。 对于 cat 的任何非 FAILED 的退出状态，baz 将会启动。spring-doc.cadn.net.cn

使用 Spring Cloud Data Flow 仪表板创建相同的“带通配符的转换”将如下图所示：spring-doc.cadn.net.cn

Composed Task Basic Transition with Wildcard
图4. 使用通配符的基本转换

29.2.3. 带后续条件执行的转换

只要不使用通配符，转换后面就可以跟一个条件执行，如下例所示：spring-doc.cadn.net.cn

task create my-transition-conditional-execution-task --definition "foo 'FAILED' -> bar 'UNKNOWN' -> baz && qux && quux"

在前面的示例中，foo 将会启动；如果其退出状态为 FAILED，则任务 bar 将会启动。 如果 foo 的退出状态为 UNKNOWN，则 baz 将会启动。 对于 foo 的任何退出状态（除 FAILED 或 UNKNOWN 外），qux 将会启动，并在成功完成后启动 quux。spring-doc.cadn.net.cn

使用 Spring Cloud Data Flow 仪表板创建相同的“带条件执行的转换”将类似于下图所示：spring-doc.cadn.net.cn

Composed Task Transition with Conditional Execution
图5. 带条件执行的转换
在此图中，虚线（转换）将 foo 应用程序连接到目标应用程序，而实线则连接 foo、qux 和 quux 之间的条件执行。

29.3. 拆分执行

拆分（Splits）允许在一个组合任务中并行运行多个任务。 它通过使用尖括号（<>）将需要并行执行的任务和流程分组来表示。 这些任务和流程之间使用双竖线 || 符号分隔，如下例所示：spring-doc.cadn.net.cn

task create my-split-task --definition "<foo || bar || baz>"

前面的示例并行启动了任务 foo、bar 和 baz。spring-doc.cadn.net.cn

使用 Spring Cloud Data Flow 仪表板创建相同的“拆分执行”将类似于下图所示：spring-doc.cadn.net.cn

Composed Task Split
图6. 拆分

使用任务 DSL，您还可以按顺序运行多个拆分组，如下例所示：spring-doc.cadn.net.cn

task create my-split-task --definition "<foo || bar || baz> && <qux || quux>"

在上面的示例中，foo、bar 和 baz 任务被并行启动。 一旦它们全部完成，qux 和 quux 任务将被并行启动。 一旦这些任务完成，组合任务即结束。 然而，如果 foo、bar 或 baz 失败，则包含 qux 和 quux 的分支将不会启动。spring-doc.cadn.net.cn

使用 Spring Cloud Data Flow 仪表板创建相同的“带多个组的拆分”将如下图所示：spring-doc.cadn.net.cn

Composed Task Split
图7. Split作为条件执行的一部分

请注意，当连接两个连续的分支时，设计器会插入一个 SYNC 控制节点。spring-doc.cadn.net.cn

在拆分（split）中使用的任务不应设置其 ExitMessage。ExitMessage 的设置仅用于转换（transitions）。

29.3.1. 包含条件执行的拆分

拆分也可以在尖括号内包含条件执行，如下例所示：spring-doc.cadn.net.cn

task create my-split-task --definition "<foo && bar || baz>"

在前面的示例中，我们看到 foo 和 baz 是并行启动的。 然而，bar 会等到 foo 成功完成后才启动。spring-doc.cadn.net.cn

使用 Spring Cloud Data Flow 仪表板创建相同的“split containing conditional execution”如下图所示：spring-doc.cadn.net.cn

Composed Task Split With Conditional Execution
图8. 带条件执行的拆分

29.3.2. 为拆分任务设置合适的线程数量

每个拆分中包含的每个子任务都需要一个线程才能运行。要正确设置此参数，您需要查看您的图表并找到具有最多子任务的拆分。该拆分中的子任务数量即为您所需的线程数。 To set the thread count, use the split-thread-core-pool-size property (defaults to 1). So, for example, a definition such as <AAA || BBB || CCC> && <DDD || EEE> requires a split-thread-core-pool-size of 3. 这是因为最大的拆分包含三个子任务。如果计数为二，则意味着 AAA 和 BBB 将并行运行，但 CCC 将等待 AAA 或 BBB 完成后方可运行。 随后 DDD 和 EEE 将并行运行。spring-doc.cadn.net.cn

30. 从流中启动任务

您可以使用 task-launcher-dataflow sink 从流启动任务。 sink 连接到 Data Flow 服务器，并使用其 REST API 启动任何已定义的任务。 sink 接受一个表示 task launch request 的 JSON 负载，其中提供要启动的任务名称，并可包含命令行参数和部署属性。spring-doc.cadn.net.cn

app-starters-task-launch-request-common 组件与 Spring Cloud Stream 函数组合结合使用，可将任何源或处理器的输出转换为任务启动请求。spring-doc.cadn.net.cn

添加对 app-starters-task-launch-request-common 的依赖会自动配置一个 java.util.function.Function 实现，并通过 Spring Cloud Function 注册为 taskLaunchRequest。spring-doc.cadn.net.cn

例如，你可以从 time 源开始，添加以下依赖项，进行构建，并将其注册为自定义源。在本例中，我们称之为 time-tlr：spring-doc.cadn.net.cn

<dependency>
    <groupId>org.springframework.cloud.stream.app</groupId>
    <artifactId>app-starters-task-launch-request-common</artifactId>
</dependency>
Spring Cloud Stream Initializr 为创建流式应用程序提供了一个绝佳的起点。

接下来，注册 task-launcher-dataflow 接收器并创建一个任务（我们使用提供的 timestamp 任务）：spring-doc.cadn.net.cn

stream create --name task-every-minute --definition "time-tlr --trigger.fixed-delay=60 --spring.cloud.stream.function.definition=taskLaunchRequest --task.launch.request.task-name=timestamp-task | task-launcher-dataflow" --deploy

前述流每分钟生成一个任务启动请求。该请求提供了要启动的任务名称：{"name":"timestamp-task"}。spring-doc.cadn.net.cn

以下流定义展示了命令行参数的使用。它会生成类似 {"args":["foo=bar","time=12/03/18 17:44:12"],"deploymentProps":{},"name":"timestamp-task"} 的消息，以向任务提供命令行参数：spring-doc.cadn.net.cn

stream create --name task-every-second --definition "time-tlr --spring.cloud.stream.function.definition=taskLaunchRequest --task.launch.request.task-name=timestamp-task --task.launch.request.args=foo=bar --task.launch.request.arg-expressions=time=payload | task-launcher-dataflow" --deploy

注意此处使用了 SpEL 表达式，将每条消息的有效载荷映射到 time 命令行参数，同时还包含一个静态参数（foo=bar）。spring-doc.cadn.net.cn

然后，您可以使用 shell 命令 task execution list 查看任务执行列表，如下例所示（包含其输出）：spring-doc.cadn.net.cn

dataflow:>task execution list
╔════════════════════╤══╤════════════════════════════╤════════════════════════════╤═════════╗
║     Task Name      │ID│         Start Time         │          End Time          │Exit Code║
╠════════════════════╪══╪════════════════════════════╪════════════════════════════╪═════════╣
║timestamp-task_26176│4 │Tue May 02 12:13:49 EDT 2017│Tue May 02 12:13:49 EDT 2017│0        ║
║timestamp-task_32996│3 │Tue May 02 12:12:49 EDT 2017│Tue May 02 12:12:49 EDT 2017│0        ║
║timestamp-task_58971│2 │Tue May 02 12:11:50 EDT 2017│Tue May 02 12:11:50 EDT 2017│0        ║
║timestamp-task_13467│1 │Tue May 02 12:10:50 EDT 2017│Tue May 02 12:10:50 EDT 2017│0        ║
╚════════════════════╧══╧════════════════════════════╧════════════════════════════╧═════════╝

在本示例中，我们展示了如何使用 time 源以固定速率启动任务。 此模式可应用于任何源，以响应任意事件来启动任务。spring-doc.cadn.net.cn

30.1. 从流中启动组合任务

可以使用 task-launcher-dataflow 接收器来启动组合任务，如此处所述。 由于我们直接使用 ComposedTaskRunner，因此在创建用于启动组合任务的流之前，需要先为组合任务运行器本身以及各个组合任务设置好任务定义。 假设我们要创建如下所示的组合任务定义：AAA && BBB。 第一步是创建任务定义，如下例所示：spring-doc.cadn.net.cn

task create composed-task-runner --definition "composed-task-runner"
task create AAA --definition "timestamp"
task create BBB --definition "timestamp"
ComposedTaskRunner 的发布版本可在此处找到。

现在，我们已经准备好用于组合任务定义所需的任务定义，接下来需要创建一个流来启动 ComposedTaskRunner。 因此，在这种情况下，我们创建一个包含以下内容的流：spring-doc.cadn.net.cn

该流应类似于以下内容：spring-doc.cadn.net.cn

stream create ctr-stream --definition "time --fixed-delay=30 --task.launch.request.task-name=composed-task-launcher --task.launch.request.args=--graph=AAA&&BBB,--increment-instance-enabled=true | task-launcher-dataflow"

目前，我们专注于启动ComposedTaskRunner所需的配置：spring-doc.cadn.net.cn

  • graph：这是将由 ComposedTaskRunner 执行的图。 在此例中，其值为 AAA&&BBB。spring-doc.cadn.net.cn

  • increment-instance-enabled：此属性使每次执行 ComposedTaskRunner 都具有唯一性。 ComposedTaskRunner 是使用 Spring Batch 构建的。 因此，我们希望每次启动 ComposedTaskRunner 时都创建一个新的作业实例（Job Instance）。 为此，我们将 increment-instance-enabled 设置为 true。spring-doc.cadn.net.cn

31. 与任务共享 Spring Cloud Data Flow 的数据存储

正如任务（Tasks）文档中所讨论的，Spring Cloud Data Flow 允许您查看 Spring Cloud Task 应用程序的执行情况。因此，在本节中，我们将讨论任务应用程序与 Spring Cloud Data Flow 之间共享任务执行信息所需的内容。spring-doc.cadn.net.cn

31.1. 一个通用的 DataStore 依赖

Spring Cloud Data Flow 开箱即用地支持多种数据库， 因此通常您只需声明 spring_datasource_* 环境变量， 即可指定 Spring Cloud Data Flow 所需使用的数据存储。 无论您为 Spring Cloud Data Flow 选择哪种数据库，请确保您的任务在其 pom.xml 或 gradle.build 文件中 也包含了相应的数据库依赖。如果任务应用程序中缺少 Spring Cloud Data Flow 所使用的数据库依赖， 该任务将失败，并且任务执行记录不会被保存。spring-doc.cadn.net.cn

31.2. 一个通用的数据存储

Spring Cloud Data Flow 与您的任务应用程序必须访问同一个数据存储实例。 这样，任务应用程序记录的任务执行信息才能被 Spring Cloud Data Flow 读取，并在 Shell 和仪表板视图中列出。 此外，任务应用程序必须对 Spring Cloud Data Flow 所使用的任务数据表具有读写权限。spring-doc.cadn.net.cn

了解了任务应用与 Spring Cloud Data Flow 之间的数据源依赖关系后，您现在可以查看如何在各种任务编排场景中应用这些知识。spring-doc.cadn.net.cn

31.2.1. 简单任务启动

当从 Spring Cloud Data Flow 启动任务时，Data Flow 会将其数据源属性（spring.datasource.url、spring.datasource.driverClassName、spring.datasource.username、spring.datasource.password）添加到所启动任务的应用程序属性中。因此，任务应用程序会将其任务执行信息记录到 Spring Cloud Data Flow 的存储库中。spring-doc.cadn.net.cn

31.2.2. 组合任务运行器

Spring Cloud Data Flow 允许您创建一个有向图，其中图中的每个节点都是一个任务应用。这是通过组合任务运行器实现的。 在这种情况下，适用于简单任务启动或任务Starters接收器的规则也同样适用于组合任务运行器。 所有子应用也必须能够访问组合任务运行器所使用的数据存储。 此外，所有子应用在其 pom.xml 或 gradle.build 文件中必须包含与组合任务运行器相同的数据库依赖项。spring-doc.cadn.net.cn

31.2.3. 从 Spring Cloud Data Flow 外部启动任务

你可以使用其他方法（例如调度器）来启动 Spring Cloud Task 应用程序，同时仍然在 Spring Cloud Data Flow 中跟踪任务的执行情况。 前提是这些任务应用程序遵循此处和此处所规定的规则。spring-doc.cadn.net.cn

如果你想使用 Spring Cloud Data Flow 来查看你的 Spring Batch 作业，请确保你的批处理应用程序使用了 @EnableTask 注解，并遵循此处和此处所列举的规则。 更多信息请参见这里。

32. 任务调度

Spring Cloud Data Flow 允许您使用 cron 表达式来调度任务的执行。 您可以通过 RESTful API 或 Spring Cloud Data Flow 用户界面创建调度计划。spring-doc.cadn.net.cn

32.1. 调度器

Spring Cloud Data Flow 通过云平台上可用的调度代理来安排其任务的执行。 在使用 Cloud Foundry 平台时，Spring Cloud Data Flow 使用 PCF Scheduler。 在使用 Kubernetes 时，将使用 CronJob。spring-doc.cadn.net.cn

计划任务不支持持续部署功能。在 Spring Cloud Data Flow 中，对任务定义的应用程序版本或属性所做的任何更改都不会影响已调度的任务。
Scheduler Architecture Overview
图9. 架构概览

32.2. 启用调度

默认情况下，Spring Cloud Data Flow 会禁用调度功能。要启用调度功能，请将以下功能属性设置为 true：spring-doc.cadn.net.cn

32.3. 调度的生命周期

调度的生命周期包含三个部分：spring-doc.cadn.net.cn

32.3.1. 调度任务执行

你可以通过以下方式调度任务执行：spring-doc.cadn.net.cn

32.3.2. 调度任务

要使用 Shell 调度任务，请使用 task schedule create 命令创建调度，如下例所示：spring-doc.cadn.net.cn

dataflow:>task schedule create --definitionName mytask --name mytaskschedule --expression '*/1 * * * *'
Created schedule 'mytaskschedule'

在前面的示例中，我们为名为 mytaskschedule 的任务定义创建了一个名为 mytask 的调度计划。该调度计划每分钟启动一次 mytask。spring-doc.cadn.net.cn

如果使用 Cloud Foundry，上面的 cron 表达式应为：*/1 * ? * *。这是因为 Cloud Foundry 使用的是 Quartz 的 cron 表达式格式。
计划名称的最大长度

计划名称的最大字符长度取决于平台。spring-doc.cadn.net.cn

表3. 各平台计划名称的最大字符长度
Kubernetes Cloud Foundry 本地

52spring-doc.cadn.net.cn

63spring-doc.cadn.net.cn

N/Aspring-doc.cadn.net.cn

32.3.3. 删除计划任务

您可以使用以下方式删除一个调度任务：spring-doc.cadn.net.cn

要使用 Shell 删除任务调度，请使用 task schedule destroy 命令，如下例所示：spring-doc.cadn.net.cn

dataflow:>task schedule destroy --name mytaskschedule
Deleted task schedule 'mytaskschedule'

32.3.4. 列出调度计划

您可以使用以下方式查看可用的日程安排：spring-doc.cadn.net.cn

要从命令行查看您的调度任务，请使用 task schedule list 命令，如下例所示：spring-doc.cadn.net.cn

dataflow:>task schedule list
╔══════════════════════════╤════════════════════╤════════════════════════════════════════════════════╗
║      Schedule Name       │Task Definition Name│                     Properties                     ║
╠══════════════════════════╪════════════════════╪════════════════════════════════════════════════════╣
║mytaskschedule            │mytask              │spring.cloud.scheduler.cron.expression = */1 * * * *║
╚══════════════════════════╧════════════════════╧════════════════════════════════════════════════════╝
使用 Spring Cloud Data Flow UI 创建、删除和列出调度任务的说明请点击此处。

33. 持续部署

随着任务应用程序的演进，您希望将更新部署到生产环境。本节将介绍 Spring Cloud Data Flow 在更新任务应用程序方面所提供的功能。spring-doc.cadn.net.cn

当一个任务应用被注册时（参见注册任务应用），会为其关联一个版本。一个任务应用可以关联多个版本，其中一个是默认版本。下图展示了一个关联了多个版本的应用（参见时间戳条目）。spring-doc.cadn.net.cn

Task Application Versions

应用程序的版本是通过注册多个具有相同名称和坐标（除了版本号之外）的应用程序来管理的。例如，如果你使用以下值注册一个应用程序，你将得到一个注册了两个版本（2.1.0.RELEASE 和 2.1.1.RELEASE）的应用程序：spring-doc.cadn.net.cn

除了拥有多个版本外，Spring Cloud Data Flow 还需要知道在下次启动时应运行哪个版本。这通过将某个版本设置为默认版本来指定。任务应用程序中被配置为默认版本的那个版本，将在下次启动请求时被运行。您可以在 UI 中查看哪个版本是默认版本，如下图所示：spring-doc.cadn.net.cn

Task Application Default Version

33.1. 任务启动生命周期

在 Spring Cloud Data Flow 的早期版本中，当收到启动任务的请求时，Spring Cloud Data Flow 会（如果需要）部署应用程序并运行它。如果应用程序运行在一个无需每次重新部署的平台上（例如 CloudFoundry），则会使用先前已部署的应用程序。此流程在 2.3 版本中发生了变化。下图展示了现在收到任务启动请求时所发生的情况：spring-doc.cadn.net.cn

Flow For Launching A Task

在上图中需要考虑三个主要流程。第一个是首次启动，或者在没有任何更改的情况下启动。另外两个分别是：当存在更改但应用程序当前未运行时的启动，以及当存在更改且应用程序正在运行时的启动。我们首先来看没有更改的情况下的流程。spring-doc.cadn.net.cn

33.1.1. 启动一个无变更的任务

  1. 一个启动请求进入 Data Flow。Data Flow 判定无需升级，因为自上次执行以来没有任何内容发生变化（属性、部署属性或版本均未更改）。spring-doc.cadn.net.cn

  1. 在会缓存已部署构件的平台上（例如撰写本文时的 CloudFoundry），Data Flow 会检查该应用程序是否先前已被部署。spring-doc.cadn.net.cn

  2. 如果应用程序需要部署，Data Flow 会部署该任务应用程序。spring-doc.cadn.net.cn

  3. Data Flow 启动应用程序。spring-doc.cadn.net.cn

此流程是默认行为，如果没有任何更改，每次请求进来时都会执行该流程。请注意，这与 Data Flow 一直以来用于启动任务的流程相同。spring-doc.cadn.net.cn

33.1.2. 启动一个包含变更但当前未运行的任务

启动任务时需要考虑的第二种流程是：当任务未运行，但任务应用程序版本、应用程序属性或部署属性中的任意一项发生变更时。在这种情况下，将执行以下流程：spring-doc.cadn.net.cn

  1. 一个启动请求进入 Data Flow。Data Flow 判定需要进行升级，因为任务应用程序版本、应用程序属性或部署属性发生了变更。spring-doc.cadn.net.cn

  2. 数据流检查当前是否有另一个任务定义实例正在运行。spring-doc.cadn.net.cn

  1. 如果没有其他正在运行的任务定义实例，则删除旧的部署。spring-doc.cadn.net.cn

  2. 在会缓存已部署构件的平台上（例如撰写本文时的 CloudFoundry），Data Flow 会检查该应用是否先前已被部署过（在此流程中，由于旧的部署已被删除，此检查结果为 false）。spring-doc.cadn.net.cn

  3. Data Flow 使用更新后的值（新应用程序版本、新合并的属性以及新合并的部署属性）来部署任务应用程序。spring-doc.cadn.net.cn

  4. Data Flow 启动应用程序。spring-doc.cadn.net.cn

这种流程从根本上实现了 Spring Cloud Data Flow 的持续部署。spring-doc.cadn.net.cn

33.1.3. 在另一个实例正在运行时启动一个带有更改的任务

最后一种主要流程是：当一个启动请求发送到 Spring Cloud Data Flow 以执行升级，但该任务定义当前正在运行。在这种情况下，由于需要先删除当前应用程序，启动操作会被阻塞。在某些平台（例如撰写本文时的 Cloud Foundry）上，删除应用程序会导致所有当前正在运行的应用程序被关闭。此功能可防止这种情况发生。以下流程描述了当一个任务发生变更而另一个实例仍在运行时所发生的情况：spring-doc.cadn.net.cn

  1. 一个启动请求进入 Data Flow。Data Flow 判定需要进行升级，因为任务应用程序版本、应用程序属性或部署属性发生了变更。spring-doc.cadn.net.cn

  2. 数据流检查当前是否有另一个任务定义实例正在运行。spring-doc.cadn.net.cn

  3. 由于该任务定义的其他实例正在运行，Data Flow 阻止了此次启动。spring-doc.cadn.net.cn

由于需要删除当前正在运行的任务，任何在请求时要求升级正在运行的任务定义的启动都将被阻止执行。