# Launching Tasks from a Stream

### 1. 通过 Stream 启动 Task <a href="#spring-cloud-dataflow-launch-tasks-from-stream" id="spring-cloud-dataflow-launch-tasks-from-stream"></a>

您可以使用 [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 ，以及命令行参数和部署属性。

&#x20;[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
<dependency>
    <groupId>org.springframework.cloud.stream.app</groupId>
    <artifactId>app-starters-task-launch-request-common</artifactId>
</dependency>
```

{% 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`。

#### &#x20;<a href="#launching_a_composed_task_from_a_stream" id="launching_a_composed_task_from_a_stream"></a>
