> For the complete documentation index, see [llms.txt](https://cxz.gitbook.io/spring-cloud-date-flow/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://cxz.gitbook.io/spring-cloud-date-flow/tasks/untitled-5.md).
# Launching Tasks from a Stream
### 1. 通过 Stream 启动 Task
您可以使用 [tasklauncher-dataflow ](https://github.com/spring-cloud-stream-app-starters/tasklauncher-dataflow/blob/master/spring-cloud-starter-stream-sink-task-launcher-dataflow/README.adoc) Sink 从 Stream 中启动任务。Sink 连接 Data Flow server 并使用 REST API 接口启动已定义好的 Task。`task launch request（任务启动请求）`接受一个 [JSON payload](https://github.com/spring-cloud-stream-app-starters/tasklauncher-dataflow/blob/master/spring-cloud-starter-stream-sink-task-launcher-dataflow/README.adoc#payload) ，可以指定 Task Name ，以及命令行参数和部署属性。
[app-starters-task-launch-request-common](https://github.com/spring-cloud-stream-app-starters/core/blob/master/common/app-starters-task-launch-request-common/README.adoc) 配合 Spring Cloud Stream 的 [functional composition](https://docs.spring.io/spring-cloud-stream/docs/current-snapshot/reference/htmlsingle/#_functional_composition) 功能，可以将任何 Source 和 Sink 的 **输出** 转换为 task launch request.
添加依赖 `app-starters-task-launch-request-common`，自动配置`java.util.function.Function`实现，通过 [Spring Cloud Function](https://cloud.spring.io/spring-cloud-function/) 注册为`taskLaunchRequest`。
例如，您可以从 [time](https://github.com/spring-cloud-stream-app-starters/time/tree/master/spring-cloud-starter-stream-source-time) source 开始，添加以下依赖项，重新构建，将其注册为自定义的 source 。我们将这个例子名称定义为 `time-tlr`
```markup
org.springframework.cloud.stream.app
app-starters-task-launch-request-common
```
{% hint style="info" %}
[Spring Cloud Stream Initializr](https://start-scs.cfapps.io/) 为创建Stream Application 提供了很好的起点。
{% endhint %}
接下来， [注册](https://docs.spring.io/spring-cloud-dataflow/docs/2.1.0.RELEASE/reference/htmlsingle/#applications) 一个 `tasklauncher-dataflow` Sink ，并创建一个Task （我们将使用timestamp task ）。
```bash
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 | tasklauncher-dataflow" \
--deploy
```
上述的 Stream 每分钟会生成一个 task launch request。该请求提供了启动的任务名称：`{"name":"timestamp-task"}`。
以下 Stream definition 说明了命令行参数的使用。它将生成消息，如下为任务提供命令行参数：
```bash
{
"args": [
"foo=bar",
"time=12/03/18 17:44:12"
],
"deploymentProps": {},
"name": "timestamp-task"
}
```
```bash
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 | tasklauncher-dataflow" \
--deploy
```
请注意，SpEL表达式会映射所有有效消息到`time`命令行参数，以及静态参数 `foo=bar`
接下来，您可以使用shell命令`task execution list`查看 Task 的执行列表，如下所示（包括输出结果）：
```
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` Source 以固定速率启动 Task 。此模式可以应用于任何 Source 进行启动 Task 以及响应任何事件。
## 1. 从流中启动 Composed Task
可以使用`tasklauncher-dataflow` Sink 启动 Composed Task ，如下所述。由于我们直接使用 ComposedTaskRunner，所以在创建启动 Composed Task 的 Stream 之前，我们需要为 Composed Task 运行器本身以及 Composed Task 设置 Task Defintion。假设我们创建以下 Composed Task Defintion : `AAA && BBB`。第一步是创建 Task Defintion ，如下例所示:
```bash
task create composed-task-runner --definition "composed-task-runner"
task create AAA --definition "timestamp"
task create BBB --definition "timestamp"
```
{% hint style="info" %}
`ComposedTaskRunner`可在[此处](https://github.com/spring-cloud-task-app-starters/composed-task-runner/releases)找到 Releases 版本。
{% endhint %}
现在我们已经准备好了 Composed Task Defintion 所需的 Task Defintion，我们需要创建一个启动`ComposedTaskRunner`的Stream。因此，在这种情况下，我们需要
* `time` ，Source 会定时发送启动 Task 的请求，[如上所述](https://docs.spring.io/spring-cloud-dataflow/docs/2.1.0.RELEASE/reference/htmlsingle/#spring-cloud-dataflow-launch-tasks-from-stream)。
* `tasklauncher-dataflow`，用于启动`ComposedTaskRunner`的 Sink。
该 Stream 应类似于以下内容：
```bash
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 | tasklauncher-dataflow"
```
现在，我们重点关注一下`ComposedTaskRunner`启动所需的配置：
* **graph**：这是由`ComposedTaskRunner`执行的 graph 。在这里是 `AAA&&BBB`。
* **increment-instance-enabled**：配置表示`ComposedTaskRunner`每次执行都是唯一的。 `ComposedTaskRunner`是使用 [Spring Batch](https://projects.spring.io/spring-batch/) 构建的。因此，我们希望每次启动时都有一个新的 `ComposedTaskRunner` 实例。要做到这一点，我们将`increment-instance-enabled`成为`true`。
####