流
本节将更详细地介绍如何创建 Streams，它们是 Spring Cloud Stream 应用程序的集合。它涵盖的主题包括 创建和部署 Streams。
如果您刚开始使用 Spring Cloud Data Flow，那么在深入研究之前，您可能应该阅读入门指南 本节。
17. 简介
Stream 是长期存在的 Spring Cloud Stream 应用程序的集合，它们通过消息传递中间件相互通信。 基于文本的 DSL 定义了应用程序之间的配置和数据流。虽然提供了许多应用程序来实施常见用例，但您通常会创建自定义 Spring Cloud Stream 应用程序来实现自定义业务逻辑。
Stream 的一般生命周期为：
-
注册应用程序。
-
创建 Stream Definition。
-
部署 Stream。
-
取消部署或销毁 Stream。
-
升级或回滚 Stream 中的应用程序。
要部署 Streams，必须将 Data Flow Server 配置为将部署委托给 Spring Cloud 生态系统中名为 Skipper 的新服务器。
此外，您还可以将 Skipper 配置为将应用程序部署到一个或多个 Cloud Foundry 组织和空间、Kubernetes 集群上的一个或多个命名空间或本地计算机。 在 Data Flow 中部署流时，您可以指定在部署时使用的平台。 Skipper 还为 Data Flow 提供了对已部署的流执行更新的功能。 更新流中的应用程序的方法有很多种，但最常见的示例之一是使用新的自定义业务逻辑升级处理器应用程序，同时保留现有的 source 和 sink 应用程序。
17.1. 流管道 DSL
流是使用受 Unix 启发的 Pipeline 语法定义的。
该语法使用竖线（称为 “管道”） 来连接多个命令。
Unix 中的命令获取进程的输出，并通过管道将其传输到进程的输入。
的输出反过来又发送到进程的输入。
每个符号将左侧命令的标准输出连接到右侧命令的标准输入。
数据从左到右流经管道。ls -l | grep key | lessls -lgrep keygrepless|
在 Data Flow 中，Unix 命令被 Spring Cloud Stream 应用程序取代，每个管道符号表示通过消息传递中间件（如 RabbitMQ 或 Apache Kafka）连接应用程序的输入和输出。
每个 Spring Cloud Stream 应用程序都注册在一个简单的名称下。 注册过程指定可从何处获取应用程序（例如，在 Maven 存储库或 Docker 注册表中）。 在 Data Flow 中，我们将 Spring Cloud Stream 应用程序分类为 Sources、Processors 或 Sinks。
举个简单的例子，考虑从 HTTP Source 收集数据并写入 File Sink。 使用 DSL 时，流描述为：
http | file
涉及某些处理的流将表示为：
http | filter | transform | file
可以使用 shell 的命令创建流定义，如以下示例所示：stream create
dataflow:> stream create --name httpIngest --definition "http | file"
Stream DSL 将传递到 command 选项。--definition
流定义的部署是通过 Shell 的命令完成的，如下所示：stream deploy
dataflow:> stream deploy --name ticktock
“入门”部分介绍如何启动服务器以及如何启动和使用 Spring Cloud Data Flow shell。
请注意，shell 调用 Data Flow Server 的 REST API。有关直接向服务器发出 HTTP 请求的更多信息，请参阅 REST API 指南。
在命名流定义时，请记住，流中的每个应用程序都将在平台上创建，其名称格式为 .因此，生成的应用程序名称的总长度不能超过 58 个字符。<stream name>-<app name> |
17.2. 流应用程序 DSL
您可以使用 Stream 应用程序 DSL 为每个 Spring Cloud Stream 应用程序定义自定义绑定属性。 有关更多信息，请参阅微型网站的 Stream Application DSL 部分。
17.3. 应用程序属性
每个应用程序都使用属性来自定义其行为。例如，源模块公开了一个设置，该设置允许更改数据摄取端口的默认值：httpport
dataflow:> stream create --definition "http --port=8090 | log" --name myhttpstream此属性实际上与标准 Spring Boot 属性相同。
Data Flow 添加了使用速记形式而不是 .
您还可以指定普通版本：portserver.portportserver.port
dataflow:> stream create --definition "http --server.port=8000 | log" --name myhttpstream这种速记行为将在 Stream Application Properties 一节中详细讨论。
如果已注册应用程序属性元数据，则可以在键入后在 shell 中使用 Tab 键自动补全来获取候选属性名称的列表。--
shell 为应用程序属性提供 Tab 键自动补全。shell 命令为所有受支持的属性提供了其他文档。app info --name <appName> --type <appType>
支持的 Stream 选项包括：、 和 。<appType>sourceprocessorsink |
18. 流生命周期
流的生命周期经历以下阶段：
-
寄存器流定义
-
使用定义创建流
-
部署流
-
销毁或取消部署流
-
升级或回滚信息流中的应用
Skipper 是一个服务器，可让您发现 Spring Boot 应用程序并在多个云平台上管理其生命周期。
Skipper 中的应用程序捆绑为包含应用程序资源位置、应用程序属性和部署属性的包。
您可以将 Skipper 包视为类似于 或 等工具中的包。apt-getbrew
当 Data Flow 部署 Stream 时，它会生成一个包并将其上传到 Skipper，该包表示 Stream 中的应用程序。 用于升级或回滚 Stream 中的应用程序的后续命令将传递给 Skipper。 此外，Stream 定义是从包中逆向工程的，并且 Stream 的状态也委托给 Skipper。
18.1. 注册 Stream 应用程序
您可以使用以下命令注册受版本控制的流应用程序。您必须提供唯一名称、应用程序类型和可解析为应用程序构件的 URI。
对于类型，请指定 、 或 。版本从 URI 解析。以下是一些示例：app registersourceprocessorsink
dataflow:>app register --name mysource --type source --uri maven://com.example:mysource:0.0.1
dataflow:>app register --name mysource --type source --uri maven://com.example:mysource:0.0.2
dataflow:>app register --name mysource --type source --uri maven://com.example:mysource:0.0.3
dataflow:>app list --id source:mysource
╔═══╤══════════════════╤═════════╤════╤════╗
║app│ source │processor│sink│task║
╠═══╪══════════════════╪═════════╪════╪════╣
║ │> mysource-0.0.1 <│ │ │ ║
║ │mysource-0.0.2 │ │ │ ║
║ │mysource-0.0.3 │ │ │ ║
╚═══╧══════════════════╧═════════╧════╧════╝
dataflow:>app register --name myprocessor --type processor --uri file:///Users/example/myprocessor-1.2.3.jar
dataflow:>app register --name mysink --type sink --uri https://example.com/mysink-2.0.1.jar应用程序 URI 应符合以下架构格式之一：
-
Maven 架构：
maven://<groupId>:<artifactId>[:<extension>[:<classifier>]]:<version> -
HTTP 架构：
http://<web-path>/<artifactName>-<version>.jar -
文件架构：
file:///<local-path>/<artifactName>-<version>.jar -
Docker 架构：
docker:<docker-image-path>/<imageName>:<version>
URI 部分对于版本控制流应用程序是必需的。
Skipper 使用多版本流应用程序，允许使用部署属性在运行时升级或回滚这些应用程序。<version> |
如果要注册使用 RabbitMQ Binder 构建的 和 应用程序的快照版本，可以执行以下操作：httplog
dataflow:>app register --name http --type source --uri maven://org.springframework.cloud.stream.app:http-source-rabbit:3.2.1
dataflow:>app register --name log --type sink --uri maven://org.springframework.cloud.stream.app:log-sink-rabbit:3.2.1如果要一次注册多个应用程序，可以将它们存储在属性文件中，其中键的格式为 URI，值为 URI。<type>.<name>
例如，要注册使用 RabbitMQ Binder 构建的 和 应用程序的快照版本，您可以在属性文件中包含以下内容（例如：httplogstream-apps.properties
source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:3.2.1
sink.log=maven://org.springframework.cloud.stream.app:log-sink-rabbit:3.2.1然后，要批量导入应用程序，请使用命令并在开关中提供属性文件的位置，如下所示：app import--uri
dataflow:>app import --uri file:///<YOUR_FILE_LOCATION>/stream-apps.properties使用 注册应用程序与注册 、 或 相同。
这种类型的应用程序只能在 Stream Application DSL（在 DSL 中使用双管道而不是单管道）中使用，并指示 Data Flow 不配置应用程序的 Spring Cloud Stream 绑定属性。
使用注册的应用程序不必是 Spring Cloud Stream 应用程序。它可以是任何 Spring Boot 应用程序。
有关使用此应用程序类型的更多信息，请参阅 Stream Application DSL 简介。--type appsourceprocessorsinkapp|||--type app
您可以注册相同应用程序的多个版本（例如，相同的名称和类型），但只能将一个版本设置为默认版本。 默认版本用于部署 Streams。
首次注册应用程序时，它将被标记为 default。可以使用以下命令更改默认应用程序版本：app default
dataflow:>app default --id source:mysource --version 0.0.2
dataflow:>app list --id source:mysource
╔═══╤══════════════════╤═════════╤════╤════╗
║app│ source │processor│sink│task║
╠═══╪══════════════════╪═════════╪════╪════╣
║ │mysource-0.0.1 │ │ │ ║
║ │> mysource-0.0.2 <│ │ │ ║
║ │mysource-0.0.3 │ │ │ ║
╚═══╧══════════════════╧═════════╧════╧════╝该命令列出了给定流应用程序的所有版本。app list --id <type:name>
该命令有一个可选参数，用于指定要取消注册的应用程序版本：app unregister--version
dataflow:>app unregister --name mysource --type source --version 0.0.1
dataflow:>app list --id source:mysource
╔═══╤══════════════════╤═════════╤════╤════╗
║app│ source │processor│sink│task║
╠═══╪══════════════════╪═════════╪════╪════╣
║ │> mysource-0.0.2 <│ │ │ ║
║ │mysource-0.0.3 │ │ │ ║
╚═══╧══════════════════╧═════════╧════╧════╝如果未指定，则默认版本未注册。--version
|
流中的所有应用程序都应为要部署的流设置默认版本。
否则，在部署期间，它们将被视为未注册的应用程序。
使用该命令设置默认值。 |
app default --id source:mysource --version 0.0.3
dataflow:>app list --id source:mysource
╔═══╤══════════════════╤═════════╤════╤════╗
║app│ source │processor│sink│task║
╠═══╪══════════════════╪═════════╪════╪════╣
║ │mysource-0.0.2 │ │ │ ║
║ │> mysource-0.0.3 <│ │ │ ║
╚═══╧══════════════════╧═════════╧════╧════╝这需要设置默认应用程序版本。
但是，and 命令可以使用所有 （默认和非默认） 已注册的应用程序版本。stream deploystream updatestream rollback
以下命令创建一个使用默认 mysource 版本 （0.0.3） 的流：
dataflow:>stream create foo --definition "mysource | log"然后我们可以将版本更新到 0.0.2：
dataflow:>stream update foo --properties version.mysource=0.0.2只有预先注册的应用程序才能用于 、 或 Stream。deployupdaterollback |
尝试将 更新到版本 （未注册） 失败。mysource0.0.1
18.1.1. 注册开箱即用的应用程序和任务
为方便起见，我们为所有开箱即用的流和任务应用程序提供了带有应用程序 URI（适用于 Maven 和 Docker）的静态文件。 您可以指向此文件并批量导入所有应用程序 URI。 否则，如前所述，您可以单独注册它们，也可以拥有自己的自定义属性文件，其中仅包含所需的应用程序 URI。 但是，我们建议在自定义属性文件中有一个所需应用程序 URI 的“重点”列表。
开箱即用的 Stream 应用程序
下表包含指向基于 Spring Cloud Stream 和 Spring Boot 的流应用程序的链接。dataflow.spring.io3.2.x2.7.x
| Artifact 类型（Artifact Type） | 稳定版 | SNAPSHOT 版本 |
|---|---|---|
RabbitMQ + Maven |
||
RabbitMQ + Docker |
||
Apache Kafka + Maven |
||
Apache Kafka + Docker |
默认情况下，开箱即用的应用程序的 actuator endpoints 是安全的。您可以通过设置以下属性来部署流来禁用安全性：app.*.spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration
|
在 Kubernetes 上，请参阅 Liveness and readiness probes 部分以了解如何配置 执行器端点的安全性。
开箱即用的任务应用程序
下表包含指向基于 Spring Cloud Task 和 Spring Boot 的任务应用程序的链接。dataflow.spring.io2.4.x2.7.x
| Artifact 类型（Artifact Type） | 稳定版 | SNAPSHOT 版本 |
|---|---|---|
Maven 系列 |
||
Jetty工人 |
有关可用的开箱即用流应用程序的更多信息，请参阅 Spring Cloud Stream 应用程序项目页面。
有关可用的开箱即用任务应用程序的更多信息，请参阅 timestamp-task 和 timestamp-batch 文档。
例如，如果您想批量注册使用 Kafka Binder 构建的所有开箱即用的流应用程序，可以使用以下命令：
$ dataflow:>app import --uri https://dataflow.spring.io/kafka-maven-latest或者，您可以使用 Rabbit Binder 注册所有流应用程序，如下所示：
$ dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest您还可以传递选项（默认情况下）来指示
Properties 文件的位置应在 shell 进程本身中解析。如果位置应该
从 Data Flow Server 进程中解析，请指定 .--localtrue--local false
|
当您使用 或 时，如果应用程序已注册到
提供的 name 和 type 以及 version，默认情况下，它不会被覆盖。如果要覆盖
预先存在的应用程序或坐标，包括该选项。 但请注意，下载后，应用程序可能会根据资源在本地缓存在 Data Flow 服务器上
位置。如果资源位置没有改变（即使实际资源字节可能不同），则
不会重新下载。另一方面，在使用资源时，使用恒定位置仍可能规避
缓存（如果使用版本）。 此外，如果已经部署了流并使用已注册应用程序的某个版本，则（强制）重新注册 不同的应用程序不起作用，直到再次部署流。 |
| 在某些情况下，资源在服务器端解析。在其他情况下， URI 将传递到运行时容器实例，并在其中对其进行解析。看 有关更多详细信息，请参阅每个 Data Flow Server 的特定文档。 |
18.1.2. 注册自定义应用程序
虽然 Data Flow 包括 source、processor、sink 应用程序，但您可以扩展这些应用程序或编写自定义 Spring Cloud Stream 应用程序。 您可以按照 Microsite 上的 Stream Development 指南创建自己的自定义应用程序。 创建自定义应用程序后，您可以注册它，如注册 Stream 应用程序中所述。
18.2. 创建流
Spring Cloud Data Flow Server 公开了一个完整的 RESTful API，用于管理流定义的生命周期，但最简单的使用方法是通过 Spring Cloud Data Flow shell。Getting Started 部分介绍了如何启动 shell。
新流是在流定义的帮助下创建的。这些定义是从一个简单的 DSL 构建的。例如，考虑一下如果我们运行以下 shell 命令会发生什么情况：
dataflow:> stream create --definition "time | log" --name ticktock这将定义一个名为 的流，该流基于 DSL 表达式。DSL 使用“管道”符号 （） 将源连接到接收器。ticktocktime | log|
该命令显示有关流的有用信息，如以下示例中所示（及其输出）：stream info
dataflow:>stream info ticktock
╔═══════════╤═════════════════╤═══════════╤══════════╗
║Stream Name│Stream Definition│Description│ Status ║
╠═══════════╪═════════════════╪═══════════╪══════════╣
║ticktock │time | log │ │undeployed║
╚═══════════╧═════════════════╧═══════════╧══════════╝18.2.1. Stream 应用程序属性
应用程序属性是与流中的每个应用程序关联的属性。部署应用程序时，应用程序属性将通过以下方式应用于应用程序 命令行参数或环境变量，具体取决于底层部署实现。
以可以在创建流时定义应用程序属性：
dataflow:> stream create --definition "time | log" --name ticktockshell 命令显示应用程序的公开应用程序属性。
有关公开属性的更多信息，请参阅应用程序元数据。app info --name <appName> --type <appType>
下面的清单显示了应用程序的公开属性：time
dataflow:> app info --name time --type source
Information about source application 'time':
Version: '3.2.1':
Default application version: 'true':
Resource URI: maven://org.springframework.cloud.stream.app:time-source-rabbit:3.2.1
╔══════════════════════════════╤══════════════════════════════╤══════════════════════════════╤══════════════════════════════╗
║ Option Name │ Description │ Default │ Type ║
╠══════════════════════════════╪══════════════════════════════╪══════════════════════════════╪══════════════════════════════╣
║spring.integration.poller.max-│Maximum number of messages to │<none> │java.lang.Integer ║
║messages-per-poll │poll per polling cycle. │ │ ║
║spring.integration.poller.fixe│Polling rate period. Mutually │<none> │java.time.Duration ║
║d-rate │exclusive with 'fixedDelay' │ │ ║
║ │and 'cron'. │ │ ║
║spring.integration.poller.fixe│Polling delay period. Mutually│<none> │java.time.Duration ║
║d-delay │exclusive with 'cron' and │ │ ║
║ │'fixedRate'. │ │ ║
║spring.integration.poller.rece│How long to wait for messages │1s │java.time.Duration ║
║ive-timeout │on poll. │ │ ║
║spring.integration.poller.cron│Cron expression for polling. │<none> │java.lang.String ║
║ │Mutually exclusive with │ │ ║
║ │'fixedDelay' and 'fixedRate'. │ │ ║
║spring.integration.poller.init│Polling initial delay. Applied│<none> │java.time.Duration ║
║ial-delay │for 'fixedDelay' and │ │ ║
║ │'fixedRate'; ignored for │ │ ║
║ │'cron'. │ │ ║
║time.date-format │Format for the date value. │MM/dd/yy HH:mm:ss │java.lang.String ║
╚══════════════════════════════╧══════════════════════════════╧══════════════════════════════╧══════════════════════════════╝下面的清单显示了应用程序的公开属性：log
dataflow:> app info --name log --type sink
Information about sink application 'log':
Version: '3.2.1':
Default application version: 'true':
Resource URI: maven://org.springframework.cloud.stream.app:log-sink-rabbit:3.2.1
╔══════════════════════════════╤══════════════════════════════╤══════════════════════════════╤══════════════════════════════╗
║ Option Name │ Description │ Default │ Type ║
╠══════════════════════════════╪══════════════════════════════╪══════════════════════════════╪══════════════════════════════╣
║log.name │The name of the logger to use.│<none> │java.lang.String ║
║log.level │The level at which to log │<none> │org.springframework.integratio║
║ │messages. │ │n.handler.LoggingHandler$Level║
║log.expression │A SpEL expression (against the│payload │java.lang.String ║
║ │incoming message) to evaluate │ │ ║
║ │as the logged message. │ │ ║
╚══════════════════════════════╧══════════════════════════════╧══════════════════════════════╧══════════════════════════════╝您可以在创建 时为 和 应用程序指定应用程序属性，如下所示：timelogstream
dataflow:> stream create --definition "time --fixed-delay=5 | log --level=WARN" --name ticktock请注意，在前面的示例中，为 和 应用程序定义的 and 属性是 shell 完成提供的“短格式”属性名称。
这些 “short-form” 属性名称仅适用于公开的属性。在所有其他情况下，应仅使用完全限定的属性名称。fixed-delayleveltimelog
18.2.2. 通用应用程序属性
除了通过 DSL 进行配置之外， Spring Cloud Data Flow 还提供了一种机制，用于将公共属性设置为所有
它启动的流式处理应用程序。
这可以通过在启动时添加前缀为 的属性来完成
服务器。
执行此操作时，服务器会将所有属性（不带前缀）传递给它启动的实例。spring.cloud.dataflow.applicationProperties.stream
例如，所有启动的应用程序都可以配置为使用特定的 Kafka 代理，方法是启动 数据流服务器具有以下选项：
--spring.cloud.dataflow.applicationProperties.stream.spring.cloud.stream.kafka.binder.brokers=192.168.1.100:9092
--spring.cloud.dataflow.applicationProperties.stream.spring.cloud.stream.kafka.binder.zkNodes=192.168.1.100:2181这样做会导致 and 属性
传递给所有启动的应用程序。spring.cloud.stream.kafka.binder.brokersspring.cloud.stream.kafka.binder.zkNodes
使用此机制配置的属性的优先级低于流部署属性。
如果在流部署时指定了具有相同键的属性（例如，覆盖公共属性），则会覆盖这些属性。app.http.spring.cloud.stream.kafka.binder.brokers |
18.3. 部署 Stream
本节介绍当 Spring Cloud Data Flow 服务器负责部署 Stream 时，如何部署 Stream。它涵盖了使用 Skipper 服务部署 Streams 和升级。有关如何设置部署属性的描述适用于 Stream 部署的两种方法。
考虑流定义：ticktock
dataflow:> stream create --definition "time | log" --name ticktock要部署流，请使用以下 shell 命令：
dataflow:> stream deploy --name ticktock数据流服务器将 和 应用程序的解析和部署委托给 Skipper。timelog
该命令显示有关流的有用信息，包括部署属性：stream info
dataflow:>stream info --name ticktock
╔═══════════╤═════════════════╤═════════╗
║Stream Name│Stream Definition│ Status ║
╠═══════════╪═════════════════╪═════════╣
║ticktock │time | log │deploying║
╚═══════════╧═════════════════╧═════════╝
Stream Deployment properties: {
"log" : {
"resource" : "maven://org.springframework.cloud.stream.app:log-sink-rabbit",
"spring.cloud.deployer.group" : "ticktock",
"version" : "2.0.1.RELEASE"
},
"time" : {
"resource" : "maven://org.springframework.cloud.stream.app:time-source-rabbit",
"spring.cloud.deployer.group" : "ticktock",
"version" : "2.0.1.RELEASE"
}
}该命令有一个重要的可选 command 参数 （called ）。
可以将 Skipper 配置为部署到多个平台。
Skipper 预先配置了一个名为 的平台，该平台将应用程序部署到运行 Skipper 的本地计算机。
命令行参数的默认值为 .
如果您通常部署到一个平台，则在安装 Skipper 时，您可以覆盖平台的配置。
否则，请将 to 指定为命令返回的值之一。--platformNamestream deploydefault--platformNamedefaultdefaultplatformNamestream platform-list
在前面的示例中，时间源每秒将当前时间作为消息发送一次，日志接收器使用日志框架输出该时间。
您可以对日志（具有后缀）进行尾部排序。日志文件位于 Data Flow Server 的日志输出中显示的目录中，如下面的清单所示：stdout<instance>
$ tail -f /var/folders/wn/8jxm_tbd1vj28c8vj37n900m0000gn/T/spring-cloud-dataflow-912434582726479179/ticktock-1464788481708/ticktock.log/stdout_0.log
2016-06-01 09:45:11.250 INFO 79194 --- [ kafka-binder-] log.sink : 06/01/16 09:45:11
2016-06-01 09:45:12.250 INFO 79194 --- [ kafka-binder-] log.sink : 06/01/16 09:45:12
2016-06-01 09:45:13.251 INFO 79194 --- [ kafka-binder-] log.sink : 06/01/16 09:45:13您还可以通过在创建流时传递标志，一步创建和部署流，如下所示：--deploy
dataflow:> stream create --definition "time | log" --name ticktock --deploy但是，在实际使用案例中，一步创建和部署流并不常见。
原因是，当您使用该命令时，您可以传入定义如何将应用程序映射到平台的属性（例如，要使用的容器的内存大小、要运行的每个应用程序的数量以及是否启用数据分区功能）。
Properties 还可以覆盖在创建流时设置的应用程序属性。
以下部分将详细介绍此功能。stream deploy
18.3.1. 部署属性
部署流时，您可以指定可以控制应用程序部署和配置方式的属性。有关更多信息，请参阅微型网站的 Deployment Properties 部分。
18.4. 销毁 Stream
您可以通过从 shell 发出命令来删除流，如下所示：stream destroy
dataflow:> stream destroy --name ticktock如果已部署流，则会在删除流定义之前取消部署该流。
18.5. 取消部署 Stream
通常，您希望停止流，但保留名称和定义以备将来使用。在这种情况下，您可以按名称对流进行描述：undeploy
dataflow:> stream undeploy --name ticktock
dataflow:> stream deploy --name ticktock您可以稍后发出该命令以重新启动它：deploy
dataflow:> stream deploy --name ticktock18.6. 验证流
有时，流定义中包含的应用程序在其注册中包含无效的 URI。
这可能是由于在应用程序注册时输入了无效的 URI，或者从要从中提取应用程序的存储库中删除了应用程序。
要验证流中包含的所有应用程序都是可解析的，用户可以使用以下命令：validate
dataflow:>stream validate ticktock
╔═══════════╤═════════════════╗
║Stream Name│Stream Definition║
╠═══════════╪═════════════════╣
║ticktock │time | log ║
╚═══════════╧═════════════════╝
ticktock is a valid stream.
╔═══════════╤═════════════════╗
║ App Name │Validation Status║
╠═══════════╪═════════════════╣
║source:time│valid ║
║sink:log │valid ║
╚═══════════╧═════════════════╝在前面的示例中，用户验证了他们的 ticktock 流。和 都是有效的。
现在，我们可以看到如果我们有一个流定义，其中包含一个具有无效 URI 的已注册应用程序，会发生什么情况：source:timesink:log
dataflow:>stream validate bad-ticktock
╔════════════╤═════════════════╗
║Stream Name │Stream Definition║
╠════════════╪═════════════════╣
║bad-ticktock│bad-time | log ║
╚════════════╧═════════════════╝
bad-ticktock is an invalid stream.
╔═══════════════╤═════════════════╗
║ App Name │Validation Status║
╠═══════════════╪═════════════════╣
║source:bad-time│invalid ║
║sink:log │valid ║
╚═══════════════╧═════════════════╝在这种情况下，Spring Cloud Data Flow 指出流无效，因为 URI 无效。source:bad-time
18.7. 更新流
要更新流，请使用 command，该命令将 or 作为 command 参数。
Skipper 有一个重要的新顶级前缀： .
以下命令 deploy stream （在部署时注册的版本为 ）：stream update--properties--propertiesFileversionhttp | loglog3.2.0
dataflow:> stream create --name httptest --definition "http --server.port=9000 | log"
dataflow:> stream deploy --name httptest
dataflow:>stream info httptest
╔══════════════════════════════╤══════════════════════════════╤════════════════════════════╗
║ Name │ DSL │ Status ║
╠══════════════════════════════╪══════════════════════════════╪════════════════════════════╣
║httptest │http --server.port=9000 | log │deploying ║
╚══════════════════════════════╧══════════════════════════════╧════════════════════════════╝
Stream Deployment properties: {
"log" : {
"spring.cloud.deployer.indexed" : "true",
"spring.cloud.deployer.group" : "httptest",
"maven://org.springframework.cloud.stream.app:log-sink-rabbit" : "3.2.0"
},
"http" : {
"spring.cloud.deployer.group" : "httptest",
"maven://org.springframework.cloud.stream.app:http-source-rabbit" : "3.2.0"
}
}然后，以下命令更新流以使用日志应用程序的版本。
在使用应用程序的特定版本更新流之前，我们需要确保应用程序已注册到该版本：3.2.1
dataflow:>app register --name log --type sink --uri maven://org.springframework.cloud.stream.app:log-sink-rabbit:3.2.1
Successfully registered application 'sink:log'然后我们可以更新应用程序：
dataflow:>stream update --name httptest --properties version.log=3.2.1您只能将预先注册的应用程序版本用于 、 或 流。deployupdaterollback |
要验证部署属性和更新版本，我们可以使用 ，如以下示例中所示（及其输出）：stream info
dataflow:>stream info httptest
╔══════════════════════════════╤══════════════════════════════╤════════════════════════════╗
║ Name │ DSL │ Status ║
╠══════════════════════════════╪══════════════════════════════╪════════════════════════════╣
║httptest │http --server.port=9000 | log │deploying ║
╚══════════════════════════════╧══════════════════════════════╧════════════════════════════╝
Stream Deployment properties: {
"log" : {
"spring.cloud.deployer.indexed" : "true",
"spring.cloud.deployer.count" : "1",
"spring.cloud.deployer.group" : "httptest",
"maven://org.springframework.cloud.stream.app:log-sink-rabbit" : "3.2.1"
},
"http" : {
"spring.cloud.deployer.group" : "httptest",
"maven://org.springframework.cloud.stream.app:http-source-rabbit" : "3.2.1"
}
}18.8. 强制更新 Stream
升级流时，您可以使用该选项来部署当前部署的应用程序的新实例，即使应用程序或部署属性尚未更改。
当应用程序本身在启动时（例如，从 Spring Cloud Config Server）获取配置信息时，需要此行为。
您可以使用该选项指定要强制升级的应用程序。
如果未指定任何应用程序名称，则所有应用程序都将强制升级。
您可以将 and 选项与 or 选项一起指定。--force--app-names--force--app-names--properties--propertiesFile
18.9. 流版本
Skipper 保留已部署流的历史记录。
更新 Stream 后，将有 Stream 的第二个版本。
您可以使用以下命令查询版本的历史记录：stream history --name <name-of-stream>
dataflow:>stream history --name httptest
╔═══════╤════════════════════════════╤════════╤════════════╤═══════════════╤════════════════╗
║Version│ Last updated │ Status │Package Name│Package Version│ Description ║
╠═══════╪════════════════════════════╪════════╪════════════╪═══════════════╪════════════════╣
║2 │Mon Nov 27 22:41:16 EST 2017│DEPLOYED│httptest │1.0.0 │Upgrade complete║
║1 │Mon Nov 27 22:40:41 EST 2017│DELETED │httptest │1.0.0 │Delete complete ║
╚═══════╧════════════════════════════╧════════╧════════════╧═══════════════╧════════════════╝18.10. 流清单
替换所有值后，Skipper 会保留所有应用程序、其应用程序属性及其部署属性的“清单”。 这表示部署到平台的内容的最终状态。 您可以使用以下命令查看 Stream 的任何版本的清单：
stream manifest --name <name-of-stream> --releaseVersion <optional-version>如果未指定 the，则返回最后一个版本的清单。--releaseVersion
以下示例显示了清单的用法：
dataflow:>stream manifest --name httptest使用该命令将产生以下输出：
# Source: log.yml
apiVersion: skipper.spring.io/v1
kind: SpringCloudDeployerApplication
metadata:
name: log
spec:
resource: maven://org.springframework.cloud.stream.app:log-sink-rabbit
version: 3.2.0
applicationProperties:
spring.cloud.dataflow.stream.app.label: log
spring.cloud.stream.bindings.input.group: httptest
spring.cloud.dataflow.stream.name: httptest
spring.cloud.dataflow.stream.app.type: sink
spring.cloud.stream.bindings.input.destination: httptest.http
deploymentProperties:
spring.cloud.deployer.indexed: true
spring.cloud.deployer.group: httptest
spring.cloud.deployer.count: 1
---
# Source: http.yml
apiVersion: skipper.spring.io/v1
kind: SpringCloudDeployerApplication
metadata:
name: http
spec:
resource: maven://org.springframework.cloud.stream.app:http-source-rabbit
version: 3.2.0
applicationProperties:
spring.cloud.dataflow.stream.app.label: http
spring.cloud.stream.bindings.output.producer.requiredGroups: httptest
server.port: 9000
spring.cloud.stream.bindings.output.destination: httptest.http
spring.cloud.dataflow.stream.name: httptest
spring.cloud.dataflow.stream.app.type: source
deploymentProperties:
spring.cloud.deployer.group: httptest大多数部署和应用程序属性由 Data Flow 设置，以使应用程序能够相互通信并发送带有标识标签的应用程序指标。
18.11. 回滚流
您可以使用以下命令回滚到流的先前版本：stream rollback
dataflow:>stream rollback --name httptest可选的 command 参数添加流的版本。
如果未指定，则回滚操作将转到以前的流版本。--releaseVersion
18.12. 应用程序计数
应用程序计数是系统的动态属性，用于指定应用程序的实例数。有关更多信息，请参阅微型网站的 Application Count 部分。
19. 流 DSL
本部分介绍流 DSL 简介中未涵盖的流 DSL 的其他功能。
19.1. 点击流
可以在流中的各种创建者终端节点创建 Tap。有关更多信息，请参阅微型网站的 Tapping a Stream 部分。
19.2. 在 Stream 中使用 Labels
当流由多个同名应用程序组成时，必须使用标签对它们进行限定。 有关更多信息，请参阅微型网站的 Labeling Applications 部分。
19.3. 命名目的地
您可以使用命名目标，而不是引用源或接收器应用程序。 有关更多信息，请参阅微型网站的 Named Destinations 部分。
19.4. Fan-in 和 Fan-out
通过使用命名目标，您可以支持扇入和扇出使用案例。 有关更多信息，请参阅微型网站的 Fan-in 和 Fan-out 部分。
20. 流式传输 Java DSL
您可以使用模块提供的基于 Java 的 DSL，而不是使用 shell 来创建和部署流。
有关更多信息，请参阅微型网站的 Java DSL 部分。spring-cloud-dataflow-rest-client
21. 使用多个 Binder 配置流式传输应用程序
在某些情况下，当需要连接到不同的消息传递时，流可以将其应用程序绑定到多个 Spring Cloud 流 Binders 中间件配置。在这些情况下，您应该确保使用其 Binder 正确配置应用程序 配置。例如，支持 Kafka 和 Rabbit Binders 的多 Binder 转换器是以中的处理器：
http | multibindertransform --expression=payload.toUpperCase() | log在前面的示例中，您将编写自己的应用程序。multibindertransform |
在此流中，每个应用程序都按以下方式连接到消息中间件：
-
HTTP 源将事件发送到 RabbitMQ （）。
rabbit1 -
Multi-Binder 转换处理器从 RabbitMQ （） 接收事件，并将处理后的事件发送到 Kafka （） 中。
rabbit1kafka1 -
日志接收器从 Kafka （） 接收事件。
kafka1
这里，和 是 Spring Cloud Stream 应用程序属性中给出的 Binder 名称。
基于此设置，应用程序的 Classpaths 中具有以下 Binders 并具有适当的配置：rabbit1kafka1
-
HTTP：Rabbit 绑定器
-
转换：Kafka 和 Rabbit 绑定器
-
日志：Kafka Binder
可以在应用程序本身中设置配置属性。
如果没有，则可以在部署流时通过属性传递它们：spring-cloud-streambinderdeployment
dataflow:>stream create --definition "http | multibindertransform --expression=payload.toUpperCase() | log" --name mystream
dataflow:>stream deploy mystream --properties "app.http.spring.cloud.stream.bindings.output.binder=rabbit1,app.multibindertransform.spring.cloud.stream.bindings.input.binder=rabbit1,
app.multibindertransform.spring.cloud.stream.bindings.output.binder=kafka1,app.log.spring.cloud.stream.bindings.input.binder=kafka1"您可以通过部署属性指定任何 Binder 配置属性来覆盖这些属性。
22. 功能组成
函数组合允许您将函数逻辑动态地附加到现有事件流应用程序。有关更多详细信息，请参阅微型网站的函数组成部分。
23. 功能应用
使用 Spring Cloud Stream 3.x 添加功能支持，您只需分别实现 Java Util 的 、 和 接口即可构建 和 应用程序。
有关此功能的更多信息，请参阅 SCDF 站点的 Functional Application Recipe。SourceSinkProcessorSupplierConsumerFunction
24. 示例
本章包括以下示例：
24.1. 简单的流处理
作为简单处理步骤的示例，我们可以使用以定义将 HTTP 发布的数据的有效负载转换为大写：
http | transform --expression=payload.toUpperCase() | log要创建此流，请在 shell 中输入以下命令：
dataflow:> stream create --definition "http --server.port=9000 | transform --expression=payload.toUpperCase() | log" --name mystream --deploy以下示例使用 shell 命令发布一些数据：
dataflow:> http post --target http://localhost:9000 --data "hello"前面的示例在日志中生成大写，如下所示：HELLO
2016-06-01 09:54:37.749 INFO 80083 --- [ kafka-binder-] log.sink : HELLO24.2. 有状态流处理
为了演示数据分区功能，下面的清单部署了一个以 Kafka 作为 Binder 的流：
dataflow:>stream create --name words --definition "http --server.port=9900 | splitter --expression=payload.split(' ') | log"
Created new stream 'words'
dataflow:>stream deploy words --properties "app.splitter.producer.partitionKeyExpression=payload,deployer.log.count=2"
Deployed stream 'words'
dataflow:>http post --target http://localhost:9900 --data "How much wood would a woodchuck chuck if a woodchuck could chuck wood"
> POST (text/plain;Charset=UTF-8) http://localhost:9900 How much wood would a woodchuck chuck if a woodchuck could chuck wood
> 202 ACCEPTED
dataflow:>runtime apps
╔════════════════════╤═══════════╤═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
║App Id / Instance Id│Unit Status│ No. of Instances / Attributes ║
╠════════════════════╪═══════════╪═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╣
║words.log-v1 │ deployed │ 2 ║
╟┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈╢
║ │ │ guid = 24166 ║
║ │ │ pid = 33097 ║
║ │ │ port = 24166 ║
║words.log-v1-0 │ deployed │ stderr = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461063/words.log-v1/stderr_0.log ║
║ │ │ stdout = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461063/words.log-v1/stdout_0.log ║
║ │ │ url = https://192.168.0.102:24166 ║
║ │ │working.dir = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461063/words.log-v1 ║
╟┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈╢
║ │ │ guid = 41269 ║
║ │ │ pid = 33098 ║
║ │ │ port = 41269 ║
║words.log-v1-1 │ deployed │ stderr = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461063/words.log-v1/stderr_1.log ║
║ │ │ stdout = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461063/words.log-v1/stdout_1.log ║
║ │ │ url = https://192.168.0.102:41269 ║
║ │ │working.dir = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461063/words.log-v1 ║
╟────────────────────┼───────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╢
║words.http-v1 │ deployed │ 1 ║
╟┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈╢
║ │ │ guid = 9900 ║
║ │ │ pid = 33094 ║
║ │ │ port = 9900 ║
║words.http-v1-0 │ deployed │ stderr = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461054/words.http-v1/stderr_0.log ║
║ │ │ stdout = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461054/words.http-v1/stdout_0.log ║
║ │ │ url = https://192.168.0.102:9900 ║
║ │ │working.dir = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803461054/words.http-v1 ║
╟────────────────────┼───────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╢
║words.splitter-v1 │ deployed │ 1 ║
╟┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈╢
║ │ │ guid = 33963 ║
║ │ │ pid = 33093 ║
║ │ │ port = 33963 ║
║words.splitter-v1-0 │ deployed │ stderr = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803437542/words.splitter-v1/stderr_0.log║
║ │ │ stdout = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803437542/words.splitter-v1/stdout_0.log║
║ │ │ url = https://192.168.0.102:33963 ║
║ │ │working.dir = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/words-1542803437542/words.splitter-v1 ║
╚════════════════════╧═══════════╧═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝当您查看日志时，您应该会看到以下内容：words.log-v1-0
2016-06-05 18:35:47.047 INFO 58638 --- [ kafka-binder-] log.sink : How
2016-06-05 18:35:47.066 INFO 58638 --- [ kafka-binder-] log.sink : chuck
2016-06-05 18:35:47.066 INFO 58638 --- [ kafka-binder-] log.sink : chuck当您查看日志时，您应该会看到以下内容：words.log-v1-1
2016-06-05 18:35:47.047 INFO 58639 --- [ kafka-binder-] log.sink : much
2016-06-05 18:35:47.066 INFO 58639 --- [ kafka-binder-] log.sink : wood
2016-06-05 18:35:47.066 INFO 58639 --- [ kafka-binder-] log.sink : would
2016-06-05 18:35:47.066 INFO 58639 --- [ kafka-binder-] log.sink : a
2016-06-05 18:35:47.066 INFO 58639 --- [ kafka-binder-] log.sink : woodchuck
2016-06-05 18:35:47.067 INFO 58639 --- [ kafka-binder-] log.sink : if
2016-06-05 18:35:47.067 INFO 58639 --- [ kafka-binder-] log.sink : a
2016-06-05 18:35:47.067 INFO 58639 --- [ kafka-binder-] log.sink : woodchuck
2016-06-05 18:35:47.067 INFO 58639 --- [ kafka-binder-] log.sink : could
2016-06-05 18:35:47.067 INFO 58639 --- [ kafka-binder-] log.sink : wood此示例显示，包含相同单词的有效负载拆分将路由到同一应用程序实例。
24.3. 其他 source 和 sink 应用程序类型
此示例显示了一些更复杂的内容：将源替换为其他内容。另一种受支持的源类型是 ，它接受通过 HTTP POST 请求摄取的数据。请注意，该源接受与数据流服务器不同的端口（默认为 8080）上的数据。默认情况下，端口是随机分配的。timehttphttp
要创建使用源但仍使用相同 sink 的流，我们将 Simple Stream Processing 示例中的原始命令更改为以下内容：httplog
dataflow:> stream create --definition "http | log" --name myhttpstream --deploy请注意，这一次，在我们实际发布一些数据（通过使用 shell 命令）之前，我们不会看到任何其他输出。要查看源正在侦听的随机分配的端口，请运行以下命令：http
dataflow:>runtime apps
╔══════════════════════╤═══════════╤═════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
║ App Id / Instance Id │Unit Status│ No. of Instances / Attributes ║
╠══════════════════════╪═══════════╪═════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╣
║myhttpstream.log-v1 │ deploying │ 1 ║
╟┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈╢
║ │ │ guid = 39628 ║
║ │ │ pid = 34403 ║
║ │ │ port = 39628 ║
║myhttpstream.log-v1-0 │ deploying │ stderr = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/myhttpstream-1542803867070/myhttpstream.log-v1/stderr_0.log ║
║ │ │ stdout = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/myhttpstream-1542803867070/myhttpstream.log-v1/stdout_0.log ║
║ │ │ url = https://192.168.0.102:39628 ║
║ │ │working.dir = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/myhttpstream-1542803867070/myhttpstream.log-v1 ║
╟──────────────────────┼───────────┼─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╢
║myhttpstream.http-v1 │ deploying │ 1 ║
╟┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈╢
║ │ │ guid = 52143 ║
║ │ │ pid = 34401 ║
║ │ │ port = 52143 ║
║myhttpstream.http-v1-0│ deploying │ stderr = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/myhttpstream-1542803866800/myhttpstream.http-v1/stderr_0.log║
║ │ │ stdout = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/myhttpstream-1542803866800/myhttpstream.http-v1/stdout_0.log║
║ │ │ url = https://192.168.0.102:52143 ║
║ │ │working.dir = /var/folders/js/7b_pn0t575l790x7j61slyxc0000gn/T/spring-cloud-deployer-6467595568759190742/myhttpstream-1542803866800/myhttpstream.http-v1 ║
╚══════════════════════╧═══════════╧═════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝您应该看到相应的源具有一个属性，该属性包含它正在侦听的主机和端口信息。现在，您可以发布到该 URL，如以下示例所示：httpurl
dataflow:> http post --target http://localhost:1234 --data "hello"
dataflow:> http post --target http://localhost:1234 --data "goodbye"然后，流将数据从源汇集到接收器实现的输出日志，产生类似于以下内容的输出：httplog
2016-06-01 09:50:22.121 INFO 79654 --- [ kafka-binder-] log.sink : hello
2016-06-01 09:50:26.810 INFO 79654 --- [ kafka-binder-] log.sink : goodbye我们还可以更改 sink 实现。您可以将输出通过管道传输到文件 （），hadoop （） 或任何其他可用的 sink 应用程序。您还可以定义自己的应用程序。filehdfs