应用
5. 来源
5.1. Debezium 源
基于 Debezium 引擎的变更数据捕获 （CDC） 源。
这Debezium Source允许捕获数据库更改事件并通过不同的消息绑定器流式传输这些事件，例如Apache Kafka,RabbitMQ以及所有 Spring Cloud Stream 支持代理。
| 此源可以与任何 Spring Cloud Stream 消息绑定器一起使用。 它不受 Kafka Connect 框架的限制或依赖。尽管这种方法很灵活，但它有一定的局限性。 |
支持所有 Debezium 配置属性。
只需在任何 Debezium 属性之前使用debezium.properties.前缀。
例如，要将 Debezium 的connector.class属性使用debezium.properties.connector.classsource 属性。
5.1.1. 数据库支持
这Debezium Source目前支持多个数据存储的 CDC：MySQL、PostgreSQL、MongoDB、Oracle、SQL Server、Db2、Vitess 和 Spanner 数据库。
5.1.2. 选项
按前缀分组的属性：
德贝齐姆
- debezium-native-configuration
-
<缺少文档>（属性，默认值：
<none>) - 标头格式
-
{@link ChangeEvent} 标头格式。默认为“JSON”。（DebeziumFormat，默认：
<none>，可能的值：JSON,AVRO,PROTOBUF) - 偏移提交策略
-
定义何时应将偏移量提交到偏移量存储的策略。（DebeziumOffsetCommitPolicy，默认：
<none>，可能的值：ALWAYS,PERIODIC,DEFAULT) - 有效负载格式
-
{@link 变更事件}密钥和有效负载格式。默认为“JSON”。（DebeziumFormat，默认：
<none>，可能的值：JSON,AVRO,PROTOBUF) - 性能
-
用于 debezium 配置属性的 Spring pass-trough 包装器。所有带有“debezium.properties.*”前缀的属性都是原生 Debezium 属性。（Map<String， String>，默认：
<none>)
debezium.supplier
- 复制标头
-
将更改事件标头复制到消息标头中。（布尔值，默认：
true)
事件扁平化配置
Debezium 提供了一种全面的消息格式，可以准确地详细说明有关系统中发生的更改的信息。
但是，有时这种格式可能不适合下游使用者，这可能需要对消息进行格式化，以便字段名称和值以简化的flattened结构。
要简化 Debezium 连接器生成的事件记录的格式，您可以使用 Debezium 事件扁平化消息转换。 使用讨人喜欢的配置，您可以配置简单的消息格式，如下所示：
--debezium.properties.transforms=unwrap
--debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
--debezium.properties.transforms.unwrap.drop.tombstones=false
--debezium.properties.transforms.unwrap.delete.handling.mode=rewrite
--debezium.properties.transforms.unwrap.add.fields=name,dbDebezium 偏移存储
当 Debezium 源代码运行时，它会从源代码读取信息并定期记录offsets定义了它处理了多少信息。
如果源重新启动，它将使用最后记录的偏移量来了解它应该在源信息中恢复读取的位置。
开箱即用，提供了以下偏移存储配置选项：
-
内存中
Doesn't persist the offset data but keeps it in memory. Therefore all offsets are lost on debezium source restart.
--debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore -
本地文件系统
Store the offsets in a file on the local file system (the file can be named anything and stored anywhere). Additionally, although the connector records the offsets with every source record it produces, the engine flushes the offsets to the backing store periodically (in the example below, once each minute).
--debezium.properties.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore --debezium.properties.offset.storage.file.filename=/tmp/offsets.dat (1) --debezium.properties.offset.flush.interval.ms=60000 (2)1 要存储偏移的文件的路径。在以下情况下需要 offset.storage`设置为FileOffsetBackingStore.2 尝试提交偏移量的时间间隔。默认值为 1 分钟。 -
Kafka 主题
Uses a Kafka topic to store offset data.
--debezium.properties.offset.storage=org.apache.kafka.connect.storage.KafkaOffsetBackingStore --debezium.properties.offset.storage.topic=my-kafka-offset-topic (1) --debezium.properties.offset.storage.partitions=2 (2) --debezium.properties.offset.storage.replication.factor=1 (3) --debezium.properties.offset.flush.interval.ms=60000 (4)1 要存储偏移量的 Kafka 主题的名称。在以下情况下需要 offset.storage设置为KafkaOffsetBackingStore.2 创建偏移存储主题时使用的分区数。 3 创建偏移存储主题时使用的复制因子。 4 尝试提交偏移量的时间间隔。默认值为 1 分钟。
可以实现org.apache.kafka.connect.storage.OffsetBackingStore接口，以提供绑定到自定义后端键值存储的偏移存储。
连接器属性
下表列出了每个连接器的所有可用 Debezium 属性。
这些属性可以通过在前缀debezium.properties.前缀。
5.1.3. 示例和测试
debezium 集成测试使用在本地计算机上运行的数据库夹具。利用 Testcontainers 的帮助预构建 debezium docker 数据库镜像。
要从 IDE 运行和调试测试，您需要从命令行部署所需的数据库映像。 以下说明说明介绍了如何从 Docker 镜像运行预配置的测试数据库。
MySQL
启动debezium/example-mysql在 docker 中：
docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:2.3.3.Final（可选）用mysql客户端连接到数据库并创建debezium具有所需凭据的用户： |
docker run -it --rm --name mysqlterm --link mysql --rm mysql:5.7 sh -c 'exec mysql -h"$MYSQL_PORT_3306_TCP_ADDR" -P"$MYSQL_PORT_3306_TCP_PORT" -uroot -p"$MYSQL_ENV_MYSQL_ROOT_PASSWORD"'
mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium' IDENTIFIED BY 'dbz';使用以下属性将 Debezium 源连接到 MySQL DB：
debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector (1)
debezium.properties.name=my-connector (2)
debezium.properties.topic.prefix=my-topic (2)
debezium.properties.database.server.id=85744 (2)
debezium.properties.database.user=debezium (3)
debezium.properties.database.password=dbz (3)
debezium.properties.database.hostname=localhost (3)
debezium.properties.database.port=3306 (3)
debezium.properties.schema=true (4)
debezium.properties.key.converter.schemas.enable=true (4)
debezium.properties.value.converter.schemas.enable=true (4)
debezium.properties.transforms=unwrap (5)
debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState (5)
debezium.properties.transforms.unwrap.add.fields=name,db (5)
debezium.properties.transforms.unwrap.delete.handling.mode=none (5)
debezium.properties.transforms.unwrap.drop.tombstones=true (5)
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory (6)
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore (6)| 1 | 将 Debezium 源配置为使用 MySqlConnector。 |
| 2 | 元数据用于识别和调度传入事件。 |
| 3 | 连接到运行在localhost:3306如debezium用户。 |
| 4 | 在ChangeEvent消息。 |
| 5 | 启用更改事件展平。 |
| 6 | 在多个启动之间保留的源状态。 |
您还可以运行DebeziumDatabasesIntegrationTest#mysql()使用此 MySQL 配置。
| 禁用 mysql GenericContainer 测试初始化代码。 |
PostgreSQL
从debezium/example-postgres:1.0Docker 镜像：
docker run -it --rm --name postgres -p 5432:5432 -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres debezium/example-postgres:2.3.3.Final您可以像这样连接到此服务器：
psql -U postgres -h localhost -p 5432使用以下属性将 Debezium 源连接到 PostgreSQL：
debezium.properties.connector.class=io.debezium.connector.postgresql.PostgresConnector (1)
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory (2)
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore (2)
debezium.properties.topic.prefix=my-topic (3)
debezium.properties.name=my-connector (3)
debezium.properties.database.server.id=85744 (3)
debezium.properties.database.user=postgres (4)
debezium.properties.database.password=postgres (4)
debezium.properties.database..dbname=postgres (4)
debezium.properties.database.hostname=localhost (4)
debezium.properties.database.port=5432 (4)
debezium.properties.schema=true (5)
debezium.properties.key.converter.schemas.enable=true (5)
debezium.properties.value.converter.schemas.enable=true (5)
debezium.properties.transforms=unwrap (6)
debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState (6)
debezium.properties.transforms.unwrap.add.fields=name,db (6)
debezium.properties.transforms.unwrap.delete.handling.mode=none (6)
debezium.properties.transforms.unwrap.drop.tombstones=true (6)| 1 | 配置Debezium Source使用 PostgresConnector。 |
| 2 | 配置 Debezium 引擎以使用memory商店。 |
| 3 | 元数据用于识别和调度传入事件。 |
| 4 | 连接到运行在localhost:5432如postgres用户。 |
| 5 | 在消息中包括“更改事件值”架构。 |
| 6 | 启用 Chage 事件展平。 |
您还可以运行DebeziumDatabasesIntegrationTest#postgres()使用此 postgres 配置。
| 禁用 postgres GenericContainer 测试初始化代码。 |
Mongo数据库
从debezium/example-mongodb:2.3.3.Final容器镜像：
docker run -it --rm --name mongodb -p 27017:27017 -e MONGODB_USER=debezium -e MONGODB_PASSWORD=dbz debezium/example-mongodb:2.3.3.Final初始化库存集合
docker exec -it mongodb sh -c 'bash -c /usr/local/bin/init-inventory.sh'在mongodb终端输出，搜索像host: "3f95a8a6516e:27017":
2019-01-10T13:46:10.004+0000 I COMMAND [conn1] command local.oplog.rs appName: "MongoDB Shell" command: replSetInitiate { replSetInitiate: { _id: "rs0", members: [ { _id: 0.0, host: "3f95a8a6516e:27017" } ] }, lsid: { id: UUID("5f477a16-d80d-41f2-9ab4-4ebecea46773") }, $db: "admin" } numYields:0 reslen:22 locks:{ Global: { acquireCount: { r: 36, w: 20, W: 2 }, acquireWaitCount: { W: 1 }, timeAcquiringMicros: { W: 312 } }, Database: { acquireCount: { r: 6, w: 4, W: 16 } }, Collection: { acquireCount: { r: 4, w: 2 } }, oplog: { acquireCount: { r: 2, w: 3 } } } protocol:op_msg 988ms加127.0.0.1 3f95a8a6516e进入您的/etc/hosts
使用以下属性将 Debezium 源连接到 MongoDB：
debezium.properties.connector.class=io.debezium.connector.mongodb.MongodbSourceConnector (1)
debezium.properties.topic.prefix=my-topic
debezium.properties.name=my-connector
debezium.properties.database.server.id=85744
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory (2)
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore (2)
debezium.properties.mongodb.hosts=rs0/localhost:27017 (3)
debezium.properties.topic.prefix=dbserver1 (3)
debezium.properties.mongodb.user=debezium (3)
debezium.properties.mongodb.password=dbz (3)
debezium.properties.database.whitelist=inventory (3)
debezium.properties.tasks.max=1 (4)
debezium.properties.schema=true (5)
debezium.properties.key.converter.schemas.enable=true (5)
debezium.properties.value.converter.schemas.enable=true (5)
debezium.properties.transforms=unwrap (6)
debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState (6)
debezium.properties.transforms.unwrap.add.fields=name,db (6)
debezium.properties.transforms.unwrap.delete.handling.mode=none (6)
debezium.properties.transforms.unwrap.drop.tombstones=true (6)| 1 | 配置Debezium Source使用 MongoDB 连接器。 |
| 2 | 配置 Debezium 引擎以使用memory. |
| 3 | 连接到运行在localhost:27017如debezium用户。 |
| 4 | debezium.io/docs/connectors/mongodb/#tasks |
| 5 | 在SourceRecord事件。 |
| 6 | 启用 Chnage 事件展平化。 |
您还可以运行DebeziumDatabasesIntegrationTest#mongodb()使用此 mongoDB 配置。
SQL 服务器
开始一个sqlserver从debezium/example-postgres:1.0Docker 镜像：
docker run -it --rm --name sqlserver -p 1433:1433 -e ACCEPT_EULA=Y -e MSSQL_PID=Standard -e SA_PASSWORD=Password! -e MSSQL_AGENT_ENABLED=true microsoft/mssql-server-linux:2017-CU9-GDR2使用示例数据表单填充 debezium SqlServer 教程：
wget https://raw.githubusercontent.com/debezium/debezium-examples/master/tutorial/debezium-sqlserver-init/inventory.sql
cat ./inventory.sql | docker exec -i sqlserver bash -c '/opt/mssql-tools/bin/sqlcmd -U sa -P $SA_PASSWORD'使用以下属性将 Debezium 源连接到 SQLServer：
debezium.properties.connector.class=io.debezium.connector.sqlserver.SqlServerConnector (1)
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory (2)
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore (2)
debezium.properties.topic.prefix=my-topic (3)
debezium.properties.name=my-connector (3)
debezium.properties.database.server.id=85744 (3)
debezium.properties.database.user=sa (4)
debezium.properties.database.password=Password! (4)
debezium.properties.database..dbname=testDB (4)
debezium.properties.database.hostname=localhost (4)
debezium.properties.database.port=1433 (4)| 1 | 配置Debezium Source以使用 SqlServerConnector。 |
| 2 | 配置 Debezium 引擎以使用memory状态存储。 |
| 3 | 元数据用于识别和调度传入事件。 |
| 4 | 连接到运行在localhost:1433如sa用户。 |
您还可以运行DebeziumDatabasesIntegrationTest#sqlServer()使用此 SqlServer 配置。
神谕
从 localhost 启动可访问的 Oracle，并使用 Debezium Vagrant 设置中描述的配置、用户和授权进行设置
使用 Debezium Oracle 教程填充示例数据：
wget https://raw.githubusercontent.com/debezium/debezium-examples/master/tutorial/debezium-with-oracle-jdbc/init/inventory.sql
cat ./inventory.sql | docker exec -i dbz_oracle sqlplus debezium/dbz@//localhost:1521/ORCLPDB15.2. 文件源
此应用程序轮询目录并将新文件或其内容发送到输出通道。 默认情况下，文件源将 File 的内容作为字节数组提供。 但是，可以使用 --file.supplier.mode 选项对其进行自定义：
-
ref 提供 java.io.File 引用
-
行 将逐行拆分文件并为每行发出一条新消息
-
contents 默认值。将文件的内容作为字节数组提供
使用时--file.supplier.mode=lines，您还可以提供附加选项--file.supplier.withMarkers=true.
如果设置为 true，则基础 FileSplitter 将在实际数据之前和之后发出额外的文件开始和文件结束标记消息。
这 2 个附加标记消息的有效负载类型为FileSplitter.FileMarker.如果未显式设置，则选项 withMarkers 默认为 false。
5.2.1. 选项
文件源具有以下选项：
按前缀分组的属性：
文件.consumer
- 标记-json
-
当 'fileMarkers == true' 时，指定它们是否应生成为 FileSplitter.FileMarker 对象或 JSON。（布尔值，默认：
true) - 模式
-
用于文件读取源的 FileReadingMode。值是 'ref' - File 对象，'lines' - 每行一条消息，或 'contents' - 以字节形式显示的内容。（FileReadingMode，默认：
<none>，可能的值：ref,lines,contents) - 带标记
-
设置为 true 以在数据之前/之后发出文件开始/文件结束标记消息。仅对 FileReadingMode 'lines' 有效。（布尔值，默认：
<none>)
文件.提供商
- 空时延迟
-
未检测到新文件时的延迟持续时间。（持续时间，默认：
1s) - 目录
-
要轮询新文件的目录。（文件，默认：
<none>) - 文件名模式
-
一个简单的Ant模式来匹配文件。（字符串，默认：
<none>) - 文件名-正则表达式
-
用于匹配文件的正则表达式模式。（模式，默认：
<none>) - 预防重复
-
设置为 true 以包含防止重复的 AcceptOnceFileListFilter。（布尔值，默认：
true)
元数据.store.dynamo-db
- 创建延迟
-
创建表重试之间的延迟。（整数，默认：
1) - 创建重试
-
创建表请求的重试次数。（整数，默认：
25) - 读取容量
-
表上的读取容量。（长，默认：
1) - 桌子
-
元数据的表名。（字符串，默认：
<none>) - 生存时间
-
表条目的 TTL。（整数，默认：
<none>) - 写入容量
-
表上的写入容量。（长，默认：
1)
元数据.store.jdbc
- 地区
-
在此存储中保留的消息的唯一分组标识符。（字符串，默认：
DEFAULT) - 表前缀
-
自定义表名的前缀。（字符串，默认：
<none>)
元数据.store.mongo-db
- 收集
-
元数据的 MongoDB 集合名称。（字符串，默认：
metadataStore)
metadata.store.redis
- 钥匙
-
元数据的 Redis 密钥。（字符串，默认：
<none>)
metadata.store
- 类型
-
指示要配置的元数据存储类型（默认值为“内存”）。您必须包含相应的 Spring Integration 依赖项才能使用持久存储。（StoreType，默认：
<none>，可能的值：mongodb,redis,dynamodb,jdbc,zookeeper,hazelcast,memory)
元数据.store.zookeeper
- 连接字符串
-
Zookeeper 连接字符串，格式为 HOST：PORT。（字符串，默认：
127.0.0.1:2181) - 编码
-
在 Zookeeper 中存储数据时使用的编码。（字符集，默认：
UTF-8) - 重试间隔
-
Zookeeper作的重试间隔（以毫秒为单位）。（整数，默认：
1000) - 根
-
根节点 - 存储条目是此节点的子节点。（字符串，默认：
/SpringIntegration-MetadataStore)
5.3. FTP 源
此源应用程序支持使用 FTP 协议传输文件。
文件从remote目录到local部署应用的目录。
默认情况下，源发出的消息作为字节数组提供。但是，这可能是
使用--mode选择：
-
裁判提供一个
java.io.File参考 -
线将逐行拆分文件并为每一行发出一条新消息
-
内容默认值。将文件的内容作为字节数组提供
使用时--mode=lines，您还可以提供附加选项--withMarkers=true.
如果设置为true，基础FileSplitter将在实际数据之前和之后发出额外的文件开始和文件结束标记消息。
这 2 个附加标记消息的有效负载类型为FileSplitter.FileMarker.选项withMarkers默认为false如果未显式设置。
另请参阅 MetadataStore 选项，了解可能的共享持久存储配置，用于防止重启时出现重复消息。
5.3.1. 输入
N/A（从 FTP 服务器获取文件）。
5.3.2. 输出
模式 = 内容
头：
-
Content-Type: application/octet-stream -
file_originalFile: <java.io.File> -
file_name: <file name>
有效载荷：
一个byte[]填充文件内容。
模式 = 线
头：
-
Content-Type: text/plain -
file_orginalFile: <java.io.File> -
file_name: <file name> -
correlationId: <UUID>（每行相同） -
sequenceNumber: <n> -
sequenceSize: 0（在读取文件之前不知道行数）
有效载荷：
一个String对于每行。
第一行前面有一条消息，其中有START标记有效负载。最后一行后面是一条消息，其中有END标记有效载荷。
标记的存在和格式由with-markers和markers-json性能。
模式 = ref
头：
没有。
有效载荷：
一个java.io.File对象。
5.3.3. 选项
ftp 源具有以下选项：
按前缀分组的属性：
文件.consumer
- 标记-json
-
当 'fileMarkers == true' 时，指定它们是否应生成为 FileSplitter.FileMarker 对象或 JSON。（布尔值，默认：
true) - 模式
-
用于文件读取源的 FileReadingMode。值是 'ref' - File 对象，'lines' - 每行一条消息，或 'contents' - 以字节形式显示的内容。（FileReadingMode，默认：
<none>，可能的值：ref,lines,contents) - 带标记
-
设置为 true 以在数据之前/之后发出文件开始/文件结束标记消息。仅对 FileReadingMode 'lines' 有效。（布尔值，默认：
<none>)
ftp.工厂
- 缓存会话
-
缓存会话。（布尔值，默认：
<none>) - 客户端模式
-
用于 FTP 会话的客户端模式。（ClientMode，默认：
<none>，可能的值：ACTIVE,PASSIVE) - 主机
-
服务器的主机名。（字符串，默认：
localhost) - 密码
-
用于连接到服务器的密码。（字符串，默认：
<none>) - 端口
-
服务器的端口。（整数，默认：
21) - 用户名
-
用于连接到服务器的用户名。（字符串，默认：
<none>)
ftp.supplier
- 自动创建本地目录
-
设置为 true 以创建本地目录（如果不存在）。（布尔值，默认：
true) - 空时延迟
-
未检测到新文件时的延迟持续时间。（持续时间，默认：
1s) - 删除远程文件
-
设置为 true 可在成功传输后删除远程文件。（布尔值，默认：
false) - 文件名模式
-
一种筛选器模式，用于匹配要传输的文件的名称。（字符串，默认：
<none>) - 文件名-正则表达式
-
用于匹配要传输的文件名称的过滤器正则表达式模式。（模式，默认：
<none>) - 本地目录
-
用于文件传输的本地目录。（文件，默认：
<none>) - 保留时间戳
-
设置为 true 以保留原始时间戳。（布尔值，默认：
true) - 远程目录
-
远程 FTP 目录。（字符串，默认：
/) - 远程文件分隔符
-
远程文件分隔符。（字符串，默认：
/) - tmp-文件后缀
-
传输过程中要使用的后缀。（字符串，默认：
.tmp)
元数据.store.dynamo-db
- 创建延迟
-
创建表重试之间的延迟。（整数，默认：
1) - 创建重试
-
创建表请求的重试次数。（整数，默认：
25) - 读取容量
-
表上的读取容量。（长，默认：
1) - 桌子
-
元数据的表名。（字符串，默认：
<none>) - 生存时间
-
表条目的 TTL。（整数，默认：
<none>) - 写入容量
-
表上的写入容量。（长，默认：
1)
元数据.store.jdbc
- 地区
-
在此存储中保留的消息的唯一分组标识符。（字符串，默认：
DEFAULT) - 表前缀
-
自定义表名的前缀。（字符串，默认：
<none>)
元数据.store.mongo-db
- 收集
-
元数据的 MongoDB 集合名称。（字符串，默认：
metadataStore)
metadata.store.redis
- 钥匙
-
元数据的 Redis 密钥。（字符串，默认：
<none>)
metadata.store
- 类型
-
指示要配置的元数据存储类型（默认值为“内存”）。您必须包含相应的 Spring Integration 依赖项才能使用持久存储。（StoreType，默认：
<none>，可能的值：mongodb,redis,dynamodb,jdbc,zookeeper,hazelcast,memory)
元数据.store.zookeeper
- 连接字符串
-
Zookeeper 连接字符串，格式为 HOST：PORT。（字符串，默认：
127.0.0.1:2181) - 编码
-
在 Zookeeper 中存储数据时使用的编码。（字符集，默认：
UTF-8) - 重试间隔
-
Zookeeper作的重试间隔（以毫秒为单位）。（整数，默认：
1000) - 根
-
根节点 - 存储条目是此节点的子节点。（字符串，默认：
/SpringIntegration-MetadataStore)
5.3.4. 示例
java -jar ftp_source.jar --ftp.supplier.remote-dir=foo --file.consumer.mode=lines --ftp.factory.host=ftpserver \
--ftp.factory.username=user --ftp.factory.password=pw --ftp.local-dir=/foo5.4. Http 源代码
侦听 HTTP 请求并将正文作为消息有效负载发出的源应用程序。
如果 Content-Type 匹配text/*或application/json，有效负载将是一个字符串，
否则，有效负载将是一个字节数组。
5.4.1. 有效载荷：
如果内容类型匹配text/*或application/json
-
String
如果内容类型不匹配text/*或application/json
-
byte array
5.4.2. 选项
http 源支持以下配置属性：
按前缀分组的属性：
http.cors
- 允许凭据
-
浏览器是否应包含与要注释的请求域关联的任何 cookie。（布尔值，默认：
<none>) - 允许的标头
-
可在实际请求期间使用的请求标头列表。（String[]，默认：
<none>) - 允许的来源
-
允许的来源列表，例如 https://domain1.com。（String[]，默认：
<none>)
http
- 映射请求标头
-
将要映射的标头。（String[]，默认：
<none>) - 路径模式
-
HTTP 端点路径映射。（字符串，默认：
/)
服务器
- 端口
-
服务器 HTTP 端口。（整数，默认：
8080)
5.5. JDBC 源代码
此源轮询来自 RDBMS 的数据。此源完全基于DataSourceAutoConfiguration，因此请参阅 Spring Boot JDBC 支持 了解更多信息。
5.5.1. 有效载荷
-
Map<String, Object>什么时候jdbc.split == true（默认）和List<Map<String, Object>>否则
5.5.2. 选项
jdbc 源具有以下选项：
按前缀分组的属性：
jdbc.supplier
- 最大行数
-
要处理查询的最大行数。（整数，默认：
0) - 查询
-
用于选择数据的查询。（字符串，默认：
<none>) - 分裂
-
是否将 SQL 结果拆分为单个消息。（布尔值，默认：
true) - 更新
-
要执行的 SQL 更新语句，用于将轮询的消息标记为“已见”。（字符串，默认：
<none>)
spring.datasource
- 驱动程序类名称
-
JDBC 驱动程序的完全限定名称。默认情况下，根据 URL 自动检测。（字符串，默认：
<none>) - 密码
-
数据库的登录密码。（字符串，默认：
<none>) - 网址
-
数据库的 JDBC URL。（字符串，默认：
<none>) - 用户名
-
数据库的登录用户名。（字符串，默认：
<none>)
spring.integration.轮询器
- 克罗恩
-
用于轮询的 Cron 表达式。与 'fixedDelay' 和 'fixedRate' 互斥。（字符串，默认：
<none>) - 固定延迟
-
轮询延迟期。与 'cron' 和 'fixedRate' 互斥。（持续时间，默认：
<none>) - 固定利率
-
轮询率周期。与 'fixedDelay' 和 'cron' 互斥。（持续时间，默认：
<none>) - 初始延迟
-
轮询初始延迟。申请 'fixedDelay' 和 'fixedRate';“cron”被忽略。（持续时间，默认：
<none>) - 每次轮询的最大消息数
-
每个轮询周期要轮询的最大消息数。（整数，默认：
<none>) - 接收超时
-
轮询消息的等待时间。（持续时间，默认：
1s)
另请参阅 Spring Boot 文档以获取补充DataSourceproperties 和TriggerProperties和MaxMessagesProperties用于轮询选项。
5.6. JMS 源代码
JMS 源允许从 JMS 接收消息。
5.6.1. 选项
JMS 源具有以下选项：
按前缀分组的属性：
jms.supplier
- 客户端 ID
-
持久订阅的客户端 ID。（字符串，默认：
<none>) - 目的地
-
接收消息的目标（队列或主题）。（字符串，默认：
<none>) - 消息选择器
-
消息的选择器。（字符串，默认：
<none>) - 会话交易
-
True 启用事务并选择 DefaultMessageListenerContainer，false 选择 SimpleMessageListenerContainer。（布尔值，默认：
true) - 订阅持久
-
对于持久订阅，则为 true。（布尔值，默认：
<none>) - 订阅名称
-
持久订阅或共享订阅的名称。（字符串，默认：
<none>) - 订阅共享
-
对于共享订阅，则为 true。（布尔值，默认：
<none>)
spring.jms
- jndi-name
-
连接工厂 JNDI 名称。设置后，优先于其他连接工厂自动配置。（字符串，默认：
<none>) - pub-sub-domain
-
默认目标类型是否为 topic。（布尔值，默认：
false)
spring.jms.listener
- 确认模式
-
容器的确认模式。默认情况下，侦听器通过自动确认进行交易。（AcknowledgeMode，默认：
<none>，可能的值：AUTO,CLIENT,DUPS_OK) - 自动启动
-
启动时自动启动容器。（布尔值，默认：
true) - 并发
-
最小并发使用者数。当未指定 max-concurrency 时，最小值也将用作最大值。（整数，默认：
<none>) - 最大并发
-
最大并发使用者数。（整数，默认：
<none>) - 接收超时
-
用于接听呼叫的超时。使用 -1 表示无等待接收，或使用 0 表示根本没有超时。后者仅在未在事务管理器中运行时才可行，并且通常不鼓励使用，因为它会阻止完全关闭。（持续时间，默认：
1s)
5.7. Apache Kafka 源代码
此模块使用来自 Apache Kafka 的消息。
5.7.1. 选项
kafka 源代码具有以下选项：
（有关 Spring for Apache Kafka 配置属性，请参阅 Spring Boot 文档）
按前缀分组的属性：
卡夫卡.supplier
- 已丢弃
-
是否在“RecordFilterStrategy”之后确认已丢弃的记录。（布尔值，默认：
false) - 记录过滤器
-
将“ConsumerRecord”作为根对象的“RecordFilterStrategy”的 SpEL 表达式。（表达式，默认值：
<none>) - 主题模式
-
Apache Kafka 主题模式来订阅。（模式，默认：
<none>) - 主题
-
Apache Kafka 主题订阅。（String[]，默认：
<none>)
弹簧.kafka
- 引导服务器
-
用于建立与 Kafka 集群的初始连接的 host：port 对的逗号分隔列表。除非被覆盖，否则适用于所有组件。（List<String>，默认值：
<none>) - 客户端 ID
-
发出请求时要传递给服务器的 ID。用于服务器端日志记录。（字符串，默认：
<none>) - 性能
-
生产者和使用者通用的其他属性，用于配置客户端。（Map<String， String>，默认：
<none>)
spring.kafka.consumer
- 自动提交间隔
-
如果将 'enable.auto.commit' 设置为 true，则消费者偏移量自动提交到 Kafka 的频率。（持续时间，默认：
<none>) - 自动偏移复位
-
当 Kafka 中没有初始偏移量或服务器上不再存在当前偏移量时该怎么办。（字符串，默认：
<none>) - 引导服务器
-
用于建立与 Kafka 集群的初始连接的 host：port 对的逗号分隔列表。覆盖消费者的全局属性。（List<String>，默认值：
<none>) - 客户端 ID
-
发出请求时要传递给服务器的 ID。用于服务器端日志记录。（字符串，默认：
<none>) - 启用自动提交
-
消费者的偏移量是否在后台定期提交。（布尔值，默认：
<none>) - 获取最大等待
-
如果没有足够的数据来立即满足“fetch-min-size”给出的要求，则服务器在响应获取请求之前阻止的最长时间。（持续时间，默认：
<none>) - 获取最小大小
-
服务器应为获取请求返回的最小数据量。（DataSize，默认：
<none>) - 组 ID
-
标识此使用者所属的使用者组的唯一字符串。（字符串，默认：
<none>) - 心跳间隔
-
检测信号之间的预期时间传递给消费者协调员。（持续时间，默认：
<none>) - 隔离级别
-
用于读取以事务方式写入的消息的隔离级别。（IsolationLevel，默认：
read-uncommitted，可能的值：READ_UNCOMMITTED,READ_COMMITTED) - 密钥解序列化器
-
键的反序列化器类。（Class<？>，默认：
<none>) - 最大轮询记录
-
在对 poll（） 的单次调用中返回的最大记录数。（整数，默认：
<none>) - 性能
-
用于配置客户端的其他特定于使用者的属性。（Map<String， String>，默认：
<none>) - 值反序列化程序
-
值的反序列化器类。（Class<？>，默认：
<none>)
spring.kafka.listener
- 确认计数
-
当 ackMode 为“COUNT”或“COUNT_TIME”时，偏移量提交之间的记录数。（整数，默认：
<none>) - 确认模式
-
侦听器 AckMode。请参阅 spring-kafka 文档。（AckMode，默认：
<none>，可能的值：RECORD,BATCH,TIME,COUNT,COUNT_TIME,MANUAL,MANUAL_IMMEDIATE) - 确认时间
-
当 ackMode 为“TIME”或“COUNT_TIME”时，偏移量提交之间的时间。（持续时间，默认：
<none>) - 异步确认
-
支持异步记录确认。仅当spring.kafka.listener.ack-mode为手动或手动立即模式时才适用。（布尔值，默认：
<none>) - 自动启动
-
是否自动启动容器。（布尔值，默认：
true) - 客户端 ID
-
侦听器的使用者 client.id 属性的前缀。（字符串，默认：
<none>) - 并发
-
要在侦听器容器中运行的线程数。（整数，默认：
<none>) - 轮询之间的空闲
-
Consumer.poll（Duration） 调用之间的睡眠间隔。（持续时间，默认：
0) - 空闲事件间隔
-
发布空闲使用者事件之间的时间（未收到任何数据）。（持续时间，默认：
<none>) - 空闲分区事件间隔
-
发布空闲分区使用者事件之间的时间（未收到分区数据）。（持续时间，默认：
<none>) - 立即停止
-
容器是在处理当前记录后停止，还是在处理上一次轮询中的所有记录后停止。（布尔值，默认：
false) - 日志容器配置
-
是否在初始化期间记录容器配置（INFO 级别）。（布尔值，默认：
<none>) - 缺少主题致命
-
如果代理上至少不存在一个已配置的主题，则容器是否应无法启动。（布尔值，默认：
false) - 监视器间隔
-
对无响应的消费者进行检查之间的时间。如果未指定持续时间后缀，则将使用秒。（持续时间，默认：
<none>) - 无轮询阈值
-
应用于“pollTimeout”的乘数，以确定消费者是否无响应。（浮点，默认：
<none>) - 轮询超时
-
轮询使用者时使用的超时。（持续时间，默认：
<none>) - 类型
-
侦听器类型。（类型，默认值：
single)
5.8. 负载发生器源
发送生成的数据并将其分派到流的源。
5.8.1. 选项
负载发生器源具有以下选项：
- load-generator.generate-timestamp
-
是否生成时间戳。（布尔值，默认：
false) - 加载生成器.消息计数
-
消息计数。（整数，默认：
1000) - load-generator.message-size
-
消息大小。（整数，默认：
1000) - load-generator.producers
-
生产者数量。（整数，默认：
1)
5.9. 邮件源
侦听电子邮件并将邮件正文作为邮件有效负载发出的源应用程序。
5.9.1. 选项
邮件源具有以下选项：
- mail.supplier.charset
-
byte[] 邮件到字符串转换的字符集。（字符串，默认：
UTF-8) - 邮件.提供商.delete
-
设置为 true 可在下载后删除电子邮件。（布尔值，默认：
false) - 邮件.提供商表达式
-
配置 SpEL 表达式以选择消息。（字符串，默认：
true) - 邮件.supplier.idle-imap
-
设置为 true 以使用 IdleImap 配置。（布尔值，默认：
false) - mail.supplier.java-邮件属性
-
JavaMail 属性作为名称-值对的新行分隔字符串，例如 'foo=bar\n baz=car'。（属性，默认值：
<none>) - mail.supplier.标记为已读
-
设置为 true 将电子邮件标记为已读。（布尔值，默认：
false) - 邮件.supplier.url
-
用于连接到邮件服务器的邮件连接 URL，例如“imaps://username：[email protected]：993/Inbox”。（URLName，默认：
<none>) - mail.supplier.user-flag
-
当服务器不支持 \Recent 时标记邮件的标志。（字符串，默认：
<none>)
5.10. MongoDB 源代码
此源轮询来自 MongoDB 的数据。
此来源完全基于MongoDataAutoConfiguration，因此请参阅 Spring Boot MongoDB 支持 了解更多信息。
5.10.1. 选项
mongodb 源代码具有以下选项：
按前缀分组的属性：
mongodb.supplier
- 收集
-
要查询的 MongoDB 集合。（字符串，默认：
<none>) - 查询
-
MongoDB 查询。（字符串，默认：
{ }) - 查询表达式
-
MongoDB查询DSL样式中的SpEL表达式。（表达式，默认值：
<none>) - 分裂
-
是否将查询结果拆分为单个消息。（布尔值，默认：
true) - 更新表达式
-
MongoDB 中的 SpEL 表达式更新了 DSL 样式。（表达式，默认值：
<none>)
spring.data.mongodb
- 附加主机
-
其他服务器主机。不能使用 URI 设置，或者如果未指定“host”。其他主机将使用默认的 mongo 端口 27017。如果要使用不同的端口，可以使用“host：port”语法。（List<String>，默认值：
<none>) - 身份验证数据库
-
身份验证数据库名称。（字符串，默认：
<none>) - 自动索引创建
-
是否启用自动索引创建。（布尔值，默认：
<none>) - 数据库
-
数据库名称。覆盖 URI 中的数据库。（字符串，默认：
<none>) - 字段命名策略
-
要使用的 FieldNamingStrategy 的完全限定名称。（Class<？>，默认：
<none>) - 主机
-
Mongo 服务器主机。不能使用 URI 设置。（字符串，默认：
<none>) - 密码
-
mongo 服务器的登录密码。不能使用 URI 设置。（Character[]，默认：
<none>) - 端口
-
Mongo 服务器端口。不能使用 URI 设置。（整数，默认：
<none>) - 副本集名称
-
集群所需的副本集名称。不能使用 URI 设置。（字符串，默认：
<none>) - URI
-
Mongo 数据库 URI。覆盖主机、端口、用户名和密码。（字符串，默认：
mongodb://localhost/test) - 用户名
-
mongo 服务器的登录用户。不能使用 URI 设置。（字符串，默认：
<none>) - uuid-表示
-
将 UUID 转换为 BSON 二进制值时使用的表示。（UuidRepresentation，默认：
java-legacy，可能的值：UNSPECIFIED,STANDARD,C_SHARP_LEGACY,JAVA_LEGACY,PYTHON_LEGACY)
spring.data.mongodb.gridfs
- 桶
-
GridFS 存储桶名称。（字符串，默认：
<none>) - 数据库
-
GridFS 数据库名称。（字符串，默认：
<none>)
spring.data.mongodb.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<none>) - 启用
-
是否启用 SSL 支持。除非另有说明，否则如果提供“捆绑包”，则自动启用。（布尔值，默认：
<none>)
另请参阅 Spring Boot 文档以获取更多内容MongoProperties性能。
See 和TriggerProperties用于轮询选项。
5.11. MQTT 源代码
启用从 MQTT 接收消息的源。
5.11.1. 有效载荷：
-
String如果二进制设置为false（默认） -
byte[]如果二进制设置为true
5.11.2. 选项
mqtt 源具有以下选项：
按前缀分组的属性：
MQTT
- 清洁会话
-
客户端和服务器是否应记住重新启动和重新连接的状态。（布尔值，默认：
true) - 连接超时
-
连接超时（以秒为单位）。（整数，默认：
30) - 保持活动间隔
-
ping间隔（以秒为单位）。（整数，默认：
60) - 密码
-
连接到代理时要使用的密码。（字符串，默认：
guest) - 坚持
-
'内存'或'文件'。（字符串，默认：
memory) - 持久性目录
-
持久性目录。（字符串，默认：
/tmp/paho) - SSL-属性
-
MQTT 客户端 SSL 属性。（Map<String， String>，默认：
<none>) - 网址
-
MQTT 代理的位置（逗号分隔的列表）。（String[]，默认：
[tcp://localhost:1883]) - 用户名
-
连接到代理时要使用的用户名。（字符串，默认：
guest)
mqtt.supplier
- 二元的
-
true 将有效负载保留为字节。（布尔值，默认：
false) - 字符集
-
用于将字节转换为 String 的字符集（当二进制为 false 时）。（字符串，默认：
UTF-8) - 客户端 ID
-
标识客户端。（字符串，默认：
stream.client.id.source) - QOS
-
qoS;所有主题的单个值或与主题匹配的逗号分隔列表。（Integer[]，默认：
[0]) - 主题
-
源将订阅的主题（逗号分隔）。（String[]，默认：
[stream.mqtt])
5.12. RabbitMQ 源代码
“rabbit”源允许从 RabbitMQ 接收消息。
在部署流之前，队列必须存在;它们不会自动创建。 您可以使用 RabbitMQ Web UI 轻松创建队列。
5.12.1. 输入
不适用
5.12.2. 输出
有效载荷
-
byte[]
5.12.3. 选项
兔子源有以下选项：
按前缀分组的属性：
兔子.提供商
- 启用重试
-
true 启用重试。（布尔值，默认：
false) - 初始重试间隔
-
启用重试时的初始重试间隔。（整数，默认：
1000) - 映射请求标头
-
将要映射的标头。（String[]，默认：
[STANDARD_REQUEST_HEADERS]) - 最大尝试次数
-
启用重试时的最大传递尝试次数。（整数，默认：
3) - 最大重试间隔
-
启用重试时的最大重试间隔。（整数，默认：
30000) - 自己的连接
-
如果为 true，请根据启动属性使用单独的连接。（布尔值，默认：
false) - 队列
-
源将侦听消息的队列。（String[]，默认：
<none>) - 重新排队
-
是否应将被拒绝的邮件重新排队。（布尔值，默认：
true) - 重试乘数
-
启用重试时重试回退乘数。（双精度，默认：
2) - 交易
-
通道是否已交易。（布尔值，默认：
false)
spring.rabbitmq
- 地址随机模式
-
用于打乱已配置地址的模式。（AddressShuffleMode，默认：
none，可能的值：NONE,RANDOM,INORDER) - 地址
-
客户端应连接到的地址的逗号分隔列表。设置后，将忽略主机和端口。（字符串，默认：
<none>) - 通道 rpc 超时
-
通道中 RPC 调用的延续超时。将其设置为零以永久等待。（持续时间，默认：
10m) - 连接超时
-
连接超时。将其设置为零以永久等待。（持续时间，默认：
<none>) - 主机
-
RabbitMQ 主机。如果设置了地址，则忽略。（字符串，默认：
localhost) - 密码
-
登录以对代理进行身份验证。（字符串，默认：
guest) - 端口
-
RabbitMQ 端口。如果设置了地址，则忽略。默认为 5672，如果启用了 SSL，则默认为 5671。（整数，默认：
<none>) - 发布者确认类型
-
确认使用的发布者类型。（ConfirmType，默认：
<none>，可能的值：SIMPLE,CORRELATED,NONE) - 发布者返回
-
是否启用发布商退货。（布尔值，默认：
false) - 请求通道最大值
-
客户端请求的每个连接的通道数。使用 0 表示无限制。（整数，默认：
2047) - 请求心跳
-
请求的检测信号超时;零表示无。如果未指定持续时间后缀，则将使用秒。（持续时间，默认：
<none>) - 用户名
-
登录用户以向代理进行身份验证。（字符串，默认：
guest) - 虚拟主机
-
连接到代理时要使用的虚拟主机。（字符串，默认：
<none>)
spring.rabbitmq.listener.简单
- 确认模式
-
容器的确认模式。（AcknowledgeMode，默认：
<none>，可能的值：NONE,MANUAL,AUTO) - 自动启动
-
是否在启动时自动启动容器。（布尔值，默认：
true) - 批量大小
-
批大小，表示为容器要使用的物理消息数。（整数，默认：
<none>) - 并发
-
侦听器调用程序线程的最小数量。（整数，默认：
<none>) - 启用消费者批处理
-
容器是否根据“receive-timeout”和“batch-size”创建一批消息。强制将 'de-batching-enabled' 设置为 true，以将生产者创建的批次的内容作为离散记录包含在批次中。（布尔值，默认：
false) - 已启用取消批处理
-
容器是否应将批处理消息作为离散消息呈现，还是使用批处理调用侦听器。（布尔值，默认：
true) - 默认重新排队拒绝
-
默认情况下，是否将被拒绝的投放重新排队。（布尔值，默认：
<none>) - 空闲事件间隔
-
应多久发布一次空闲容器事件。（持续时间，默认：
<none>) - 最大并发
-
侦听器调用程序线程的最大数量。（整数，默认：
<none>) - 缺少队列致命
-
如果容器声明的队列在代理上不可用，是否失败和/或如果在运行时删除一个或多个队列，是否停止容器。（布尔值，默认：
true) - 预取
-
每个使用者可以未完成的未确认消息的最大数量。（整数，默认：
<none>)
spring.rabbitmq.listener
- 类型
-
侦听器容器类型。（ContainerType，默认：
simple，可能的值：SIMPLE,DIRECT,STREAM)
另请参阅 Spring Boot 文档，了解代理连接和侦听器属性的附加属性。
关于重试的说明
使用默认的 ackMode （AUTO） 和重新排队 （true） 选项，将重试失败的消息传递
无限期。
由于兔源没有太多的加工，因此源本身失败的风险很小，除非
下游Binder由于某种原因未连接。
将 requeue 设置为 false 将导致邮件在第一次尝试时被拒绝（并可能发送到死信
Exchange/Queue（如果代理已如此配置）。
enableRetry 选项允许配置重试参数，以便可以重试失败的消息传递，并且
当重试用尽时，最终被丢弃（或死信）。
在重试间隔期间，传递线程将挂起。
重试选项包括 enableRetry、maxAttempts、initialRetryInterval、retryMultipler 和 maxRetryInterval。
永远不会重试因 MessageConversionException 而失败的消息传递;假设如果消息
第一次尝试时无法转换，后续尝试也将失败。
此类消息将被丢弃（或死信）。 |
5.12.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps您可以在此处找到相应的基于活页夹的项目。 然后，您可以 cd 到其中一个文件夹中并构建它：
$ ./mvnw clean package5.12.5. 示例
java -jar rabbit-source.jar --rabbit.queues=5.13. Amazon S3 源代码
此源应用程序支持使用 Amazon S3 协议传输文件。
文件从remote目录（S3 存储桶）到local部署应用程序的目录。
默认情况下，源发出的消息作为字节数组提供。但是，这可能是
使用--mode选择：
-
裁判提供一个
java.io.File参考 -
线将逐行拆分文件并为每一行发出一条新消息
-
内容默认值。将文件的内容作为字节数组提供
使用时--mode=lines，您还可以提供附加选项--withMarkers=true.
如果设置为true，基础FileSplitter将在实际数据之前和之后发出额外的文件开始和文件结束标记消息。
这 2 个附加标记消息的有效负载类型为FileSplitter.FileMarker.选项withMarkers默认为false如果未显式设置。
另请参阅 MetadataStore 选项，了解可能的共享持久存储配置，用于防止重启时出现重复消息。
5.13.1. 模式 = 行
头：
-
Content-Type: text/plain -
file_orginalFile: <java.io.File> -
file_name: <file name> -
correlationId: <UUID>（每行相同） -
sequenceNumber: <n> -
sequenceSize: 0（在读取文件之前不知道行数）
有效载荷：
一个String对于每行。
第一行前面有一条消息，其中有START标记有效负载。最后一行后面是一条消息，其中有END标记有效载荷。
标记的存在和格式由with-markers和markers-json性能。
5.13.2. 模式 = ref
头：
没有。
有效载荷：
一个java.io.File对象。
5.13.3. 选项
这s3-source具有以下选项：
按前缀分组的属性：
文件.consumer
- 标记-json
-
当 'fileMarkers == true' 时，指定它们是否应生成为 FileSplitter.FileMarker 对象或 JSON。（布尔值，默认：
true) - 模式
-
用于文件读取源的 FileReadingMode。值是 'ref' - File 对象，'lines' - 每行一条消息，或 'contents' - 以字节形式显示的内容。（FileReadingMode，默认：
<none>，可能的值：ref,lines,contents) - 带标记
-
设置为 true 以在数据之前/之后发出文件开始/文件结束标记消息。仅对 FileReadingMode 'lines' 有效。（布尔值，默认：
<none>)
元数据.store.dynamo-db
- 创建延迟
-
创建表重试之间的延迟。（整数，默认：
1) - 创建重试
-
创建表请求的重试次数。（整数，默认：
25) - 读取容量
-
表上的读取容量。（长，默认：
1) - 桌子
-
元数据的表名。（字符串，默认：
<none>) - 生存时间
-
表条目的 TTL。（整数，默认：
<none>) - 写入容量
-
表上的写入容量。（长，默认：
1)
元数据.store.jdbc
- 地区
-
在此存储中保留的消息的唯一分组标识符。（字符串，默认：
DEFAULT) - 表前缀
-
自定义表名的前缀。（字符串，默认：
<none>)
元数据.store.mongo-db
- 收集
-
元数据的 MongoDB 集合名称。（字符串，默认：
metadataStore)
metadata.store.redis
- 钥匙
-
元数据的 Redis 密钥。（字符串，默认：
<none>)
metadata.store
- 类型
-
指示要配置的元数据存储类型（默认值为“内存”）。您必须包含相应的 Spring Integration 依赖项才能使用持久存储。（StoreType，默认：
<none>，可能的值：mongodb,redis,dynamodb,jdbc,zookeeper,hazelcast,memory)
元数据.store.zookeeper
- 连接字符串
-
Zookeeper 连接字符串，格式为 HOST：PORT。（字符串，默认：
127.0.0.1:2181) - 编码
-
在 Zookeeper 中存储数据时使用的编码。（字符集，默认：
UTF-8) - 重试间隔
-
Zookeeper作的重试间隔（以毫秒为单位）。（整数，默认：
1000) - 根
-
根节点 - 存储条目是此节点的子节点。（字符串，默认：
/SpringIntegration-MetadataStore)
S3.提供商
- 自动创建本地目录
-
创建或不创建本地目录。（布尔值，默认：
true) - 删除远程文件
-
处理后删除或不删除远程文件。（布尔值，默认：
false) - 文件名模式
-
过滤远程文件的模式。（字符串，默认：
<none>) - 文件名-正则表达式
-
用于过滤远程文件的正则表达式。（模式，默认：
<none>) - 仅列表
-
设置为 true 可返回 s3 对象元数据，而不将文件复制到本地目录。（布尔值，默认：
false) - 本地目录
-
用于存储文件的本地目录。（文件，默认：
<none>) - 保留时间戳
-
将远程文件的时间戳传输到本地文件。（布尔值，默认：
true) - 远程目录
-
AWS S3 存储桶资源。（字符串，默认：
bucket) - 远程文件分隔符
-
远程文件分隔符。（字符串，默认：
/) - tmp-文件后缀
-
临时文件后缀。（字符串，默认：
.tmp)
spring.cloud.aws.凭证
- 访问密钥
-
要与静态提供程序一起使用的访问密钥。（字符串，默认：
<none>) - 实例配置文件
-
配置实例配置文件凭据提供程序，无需进一步配置。（布尔值，默认：
false) - 轮廓
-
AWS 配置文件。（配置文件，默认：
<none>) - 密钥
-
要与静态提供程序一起使用的密钥。（字符串，默认：
<none>)
spring.cloud.aws.region
- 实例配置文件
-
配置实例配置文件区域提供程序，无需进一步配置。（布尔值，默认：
false) - 轮廓
-
AWS 配置文件。（配置文件，默认：
<none>) - 静态的
-
<缺少文档>（字符串，默认：
<none>)
spring.cloud.aws.s3
- 启用加速模式
-
在访问 S3 时启用使用加速端点的选项。Accelerate 终端节点允许使用 Amazon CloudFront 的全球分布式边缘站点更快地传输对象。（布尔值，默认：
<none>) - 启用校验和验证
-
禁用对存储在 S3 中的对象的校验和进行验证的选项。（布尔值，默认：
<none>) - 已启用分块编码
-
在签署 {@link software.amazon.awssdk.services.s3.model.PutObjectRequest} 和 {@link software.amazon.awssdk.services.s3.model.UploadPartRequest} 的请求负载时启用使用分块编码的选项。（布尔值，默认：
<none>) - 跨区域启用
-
启用跨区域存储桶访问。（布尔值，默认：
<none>) - 端点
-
覆盖默认终结点。（URI，默认：
<none>) - 路径样式访问启用
-
启用使用路径样式访问而不是 DNS 样式访问的选项。DNS 样式的访问是首选，因为它在访问 S3 时可以实现更好的负载平衡。（布尔值，默认：
<none>) - 地区
-
覆盖默认区域。（字符串，默认：
<none>) - 使用arn-region-enabled
-
如果 S3 资源 ARN 作为 S3作的目标传入，该作的区域与配置客户端的区域不同，则必须将此标志设置为“true”，以允许客户端对 ARN 中指定的区域进行跨区域调用，否则将引发异常。（布尔值，默认：
<none>)
5.13.4. Amazon AWS 通用选项
Amazon S3 源（与所有其他 Amazon AWS 应用程序一样）基于 Spring Cloud AWS 项目作为基础，其自动配置类由 Spring Boot 自动使用。 请参阅他们的文档，了解所需和有用的自动配置属性。
5.13.5. 示例
java -jar s3-source.jar --s3.remoteDir=/tmp/foo --file.consumer.mode=lines5.14. SFTP 源
此源应用程序支持使用 SFTP 协议传输文件。
文件从remote目录到local部署应用的目录。
默认情况下，源发出的消息作为字节数组提供。但是，这可能是
使用--mode选择：
-
裁判提供一个
java.io.File参考 -
线将逐行拆分文件并为每一行发出一条新消息
-
内容默认值。将文件的内容作为字节数组提供
使用时--mode=lines，您还可以提供附加选项--withMarkers=true.
如果设置为true，基础FileSplitter将在实际数据之前和之后发出额外的文件开始和文件结束标记消息。
这 2 个附加标记消息的有效负载类型为FileSplitter.FileMarker.选项withMarkers默认为false如果未显式设置。
看sftp-supplier以获取高级配置选项。
另请参阅 MetadataStore 选项，了解可能的共享持久存储配置，用于防止重启时出现重复消息。
5.14.1. 输入
N/A（从 SFTP 服务器获取文件）。
5.14.2. 输出
模式 = 内容
头：
-
Content-Type: application/octet-stream -
file_name: <file name> -
file_remoteFileInfo <file metadata> -
file_remoteHostPort: <host:port> -
file_remoteDirectory: <relative-path> -
file_remoteFile: <file-name> -
sftp_selectedServer: <server-key>（如果是多源）
有效载荷：
一个byte[]填充文件内容。
模式 = 线
头：
-
Content-Type: text/plain -
file_name: <file name> -
correlationId: <UUID>（每行相同） -
sequenceNumber: <n> -
sequenceSize: 0（在读取文件之前不知道行数） -
file_marker : <file marker>（如果启用了 with-markers）
有效载荷：
一个String对于每行。
第一行前面有一条消息，其中有START标记有效负载。最后一行后面是一条消息，其中有END标记有效载荷。
标记的存在和格式由with-markers和markers-json性能。
模式 = ref
头：
-
file_remoteHostPort: <host:port> -
file_remoteDirectory: <relative-path> -
file_remoteFile: <file-name> -
file_originalFile: <absolute-path-of-local-file> -
file_name <local-file-name> -
file_relativePath -
file_remoteFile: <remote-file-name> -
sftp_selectedServer: <server-key>（如果是多源）
有效载荷：
一个java.io.File对象。
5.14.3. 选项
ftp 源具有以下选项：
按前缀分组的属性：
文件.consumer
- 标记-json
-
当 'fileMarkers == true' 时，指定它们是否应生成为 FileSplitter.FileMarker 对象或 JSON。（布尔值，默认：
true) - 模式
-
用于文件读取源的 FileReadingMode。值是 'ref' - File 对象，'lines' - 每行一条消息，或 'contents' - 以字节形式显示的内容。（FileReadingMode，默认：
<none>，可能的值：ref,lines,contents) - 带标记
-
设置为 true 以在数据之前/之后发出文件开始/文件结束标记消息。仅对 FileReadingMode 'lines' 有效。（布尔值，默认：
<none>)
元数据.store.dynamo-db
- 创建延迟
-
创建表重试之间的延迟。（整数，默认：
1) - 创建重试
-
创建表请求的重试次数。（整数，默认：
25) - 读取容量
-
表上的读取容量。（长，默认：
1) - 桌子
-
元数据的表名。（字符串，默认：
<none>) - 生存时间
-
表条目的 TTL。（整数，默认：
<none>) - 写入容量
-
表上的写入容量。（长，默认：
1)
元数据.store.jdbc
- 地区
-
在此存储中保留的消息的唯一分组标识符。（字符串，默认：
DEFAULT) - 表前缀
-
自定义表名的前缀。（字符串，默认：
<none>)
元数据.store.mongo-db
- 收集
-
元数据的 MongoDB 集合名称。（字符串，默认：
metadataStore)
metadata.store.redis
- 钥匙
-
元数据的 Redis 密钥。（字符串，默认：
<none>)
metadata.store
- 类型
-
指示要配置的元数据存储类型（默认值为“内存”）。您必须包含相应的 Spring Integration 依赖项才能使用持久存储。（StoreType，默认：
<none>，可能的值：mongodb,redis,dynamodb,jdbc,zookeeper,hazelcast,memory)
元数据.store.zookeeper
- 连接字符串
-
Zookeeper 连接字符串，格式为 HOST：PORT。（字符串，默认：
127.0.0.1:2181) - 编码
-
在 Zookeeper 中存储数据时使用的编码。（字符集，默认：
UTF-8) - 重试间隔
-
Zookeeper作的重试间隔（以毫秒为单位）。（整数，默认：
1000) - 根
-
根节点 - 存储条目是此节点的子节点。（字符串，默认：
/SpringIntegration-MetadataStore)
SFTP.提供商
- 自动创建本地目录
-
设置为 true 以创建本地目录（如果不存在）。（布尔值，默认：
true) - 空时延迟
-
未检测到新文件时的延迟持续时间。（持续时间，默认：
1s) - 删除远程文件
-
设置为 true 可在成功传输后删除远程文件。（布尔值，默认：
false) - 目录
-
工厂“name.directory”对的列表。（String[]，默认：
<none>) - 工厂
-
工厂名称到工厂的映射。（Map<String， Factory>，默认值：
<none>) - 公平
-
对于多个服务器/目录的公平轮换为 True。默认情况下，这是 false，因此如果一个源有多个条目，则将在访问其他源之前收到这些条目。（布尔值，默认：
false) - 文件名模式
-
一种筛选器模式，用于匹配要传输的文件的名称。（字符串，默认：
<none>) - 文件名-正则表达式
-
用于匹配要传输的文件名称的过滤器正则表达式模式。（模式，默认：
<none>) - 仅列表
-
设置为 true 以返回不包含整个有效负载的文件元数据。（布尔值，默认：
false) - 本地目录
-
用于文件传输的本地目录。（文件，默认：
<none>) - 最大获取
-
每次轮询要获取的最大远程文件数;默认无限制。在列出文件或构建任务启动请求时不适用。（整数，默认：
<none>) - 保留时间戳
-
设置为 true 以保留原始时间戳。（布尔值，默认：
true) - 远程目录
-
远程 FTP 目录。（字符串，默认：
/) - 远程文件分隔符
-
远程文件分隔符。（字符串，默认：
/) - 将远程文件重命名为
-
成功传输后，必须将解析为新名称远程文件的 SpEL 表达式重命名为。（表达式，默认值：
<none>) - 流
-
设置为 true 以流式传输文件而不是复制到本地目录。（布尔值，默认：
false) - tmp-文件后缀
-
传输过程中要使用的后缀。（字符串，默认：
.tmp)
sftp.提供商.工厂
- 允许未知键
-
True 表示允许未知或更改的键。（布尔值，默认：
false) - 主机
-
服务器的主机名。（字符串，默认：
localhost) - 已知主机表达式
-
解析为已知主机文件位置的 SpEL 表达式。（表达式，默认值：
<none>) - 密码短语
-
用户私钥的密码。（字符串，默认：
<empty string>) - 密码
-
用于连接到服务器的密码。（字符串，默认：
<none>) - 端口
-
服务器的端口。（整数，默认：
22) - 私钥
-
用户私钥的资源位置。（资源，默认值：
<none>) - 用户名
-
用于连接到服务器的用户名。（字符串，默认：
<none>)
sftp.supplier.sort-by
- 属性
-
要排序依据的文件列表条目的属性（FILENAME、ATIME：上次访问时间、MTIME：上次修改时间）。（属性，默认：
<none>) - 迪尔
-
分拣方向（ASC 或 DESC）。（Dir，默认：
<none>)
5.14.4. 示例
java -jar sftp_source.jar --sftp.supplier.remote-dir=foo --file.mode=lines --sftp.supplier.factory.host=sftpserver \
--sftp.supplier.factory.username=user --sftp.supplier.factory.password=pw --sftp.supplier.local-dir=/foo5.15. 系统日志
系统日志源通过 UDP、TCP 或两者接收 SYSLOG 数据包。支持 RFC3164 （BSD） 和 RFC5424 格式。
5.15.1. 选项
- syslog.supplier.buffer-size
-
解码消息时使用的缓冲区大小;较大的消息将被拒绝。（整数，默认：
2048) - 系统日志.supplier.nio
-
是否使用蔚来（支持大量连接时）。（布尔值，默认：
false) - 系统日志.supplier.port
-
要侦听的端口。（整数，默认：
1514) - syslog.supplier.protocol 系统日志
-
用于 SYSLOG（tcp 或 udp）的协议。（协议，默认：
<none>，可能的值：tcp,udp,both) - syslog.supplier.反向查找
-
是否对传入套接字执行反向查找。（布尔值，默认：
false) - syslog.supplier.rfc
-
'5424' 或 '3164' - 根据 RFC 的 syslog 格式;3164 又名 'BSD' 格式。（字符串，默认：
3164) - syslog.supplier.socket-超时
-
套接字超时。（整数，默认：
0)
5.16. TCP的
这tcp源充当服务器，允许远程方连接到它并通过原始 TCP 套接字提交数据。
TCP 是一种流式处理协议，需要某种机制来构建在线路上的消息。许多解码器是 可用，默认值为与 Telnet 兼容的“CRLF”。
TCP 源应用程序生成的消息具有byte[]有效载荷。
5.16.1. 选项
按前缀分组的属性：
TCP
- 蔚来
-
是否使用蔚来。（布尔值，默认：
false) - 端口
-
要收听的端口;0 表示作系统选择端口。（整数，默认：
1234) - 反向查找
-
对远程 IP 地址执行反向 DNS 查找;如果为 false，则邮件标头中仅包含 IP 地址。（布尔值，默认：
false) - 套接字超时
-
未收到数据时关闭套接字之前的超时 （ms）。（整数，默认：
120000) - 使用直接缓冲区
-
是否使用直接缓冲区。（布尔值，默认：
false)
tcp.提供商
- 缓冲区大小
-
解码消息时使用的缓冲区大小;较大的邮件将被拒绝。（整数，默认：
2048) - 译码器
-
接收消息时要使用的解码器。（编码，默认：
<none>，可能的值：CRLF,LF,NULL,STXETX,RAW,L1,L2,L4)
5.16.2. 可用的解码器
- CRLF（默认）
-
以回车符 （0x0d） 结尾的文本，后跟换行符 （0x0a）
- 如果
-
以换行符结尾的文本 （0x0a）
- 零
-
以空字节 （0x00） 结尾的文本
- STXETX
-
前面是 STX （0x02） 并以 ETX （0x03） 结尾的文本
- 生
-
no structure - 客户端通过关闭套接字来指示完整的消息
- L1
-
前面有一个 1 字节（无符号）长度字段的数据（最多支持 255 字节）
- L2
-
数据前面有一个两个字节（无符号）长度字段（最多 216-1 字节）
- L4
-
数据前面有一个四字节（有符号）长度字段（最多 231-1 字节）
5.17. 时间源
时间源将只是每隔一段时间发出一个包含当前时间的字符串。
5.17.1. 选项
时间源具有以下选项：
- spring.integration.poller.cron
-
用于轮询的 Cron 表达式。与 'fixedDelay' 和 'fixedRate' 互斥。（字符串，默认：
<none>) - spring.integration.poller.fixed-delay
-
轮询延迟期。与 'cron' 和 'fixedRate' 互斥。（持续时间，默认：
<none>) - spring.integration.poller.固定速率
-
轮询率周期。与 'fixedDelay' 和 'cron' 互斥。（持续时间，默认：
<none>) - spring.integration.poller.initial-delay
-
轮询初始延迟。申请 'fixedDelay' 和 'fixedRate';“cron”被忽略。（持续时间，默认：
<none>) - 每次轮询spring.integration.poller.max消息
-
每个轮询周期要轮询的最大消息数。（整数，默认：
<none>) - spring.integration.poller.receive-timeout
-
轮询消息的等待时间。（持续时间，默认：
1s) - time.date-format
-
日期值的格式。（字符串，默认：
MM/dd/yy HH:mm:ss)
5.18. Twitter 消息源
重复检索过去 30 天内的直接消息（已发送和已接收），按时间倒序排序。
释放的消息被缓存（在MetadataStorecache）以防止重复。
默认情况下，内存中的SimpleMetadataStore被使用。
这twitter.message.source.count控制返回的消息数。
这spring.cloud.stream.poller属性控制消息轮询间隔。
必须与使用的 API 速率限制保持一致
5.18.1. 选项
按前缀分组的属性：
spring.integration.轮询器
- 克罗恩
-
用于轮询的 Cron 表达式。与 'fixedDelay' 和 'fixedRate' 互斥。（字符串，默认：
<none>) - 固定延迟
-
轮询延迟期。与 'cron' 和 'fixedRate' 互斥。（持续时间，默认：
<none>) - 固定利率
-
轮询率周期。与 'fixedDelay' 和 'cron' 互斥。（持续时间，默认：
<none>) - 初始延迟
-
轮询初始延迟。申请 'fixedDelay' 和 'fixedRate';“cron”被忽略。（持续时间，默认：
<none>) - 每次轮询的最大消息数
-
每个轮询周期要轮询的最大消息数。（整数，默认：
<none>) - 接收超时
-
轮询消息的等待时间。（持续时间，默认：
1s)
推特连接
- 访问Tokens
-
您的 Twitter Tokens。（字符串，默认：
<none>) - 访问Tokens密钥
-
你的 Twitter Tokens密钥。（字符串，默认：
<none>) - 消费者键
-
您的 Twitter 密钥。（字符串，默认：
<none>) - 消费者机密
-
你的 Twitter 秘密。（字符串，默认：
<none>) - 启用调试
-
启用 Twitter4J 调试模式。（布尔值，默认：
false) - 原始 JSON
-
启用缓存 Twitter API 返回的原始（原始）JSON 对象。当设置为 False 时，结果将使用 Twitter4J 的 json 表示形式。当设置为 True 时，结果将使用原始的 Twitter APISs json 表示形式。（布尔值，默认：
true)
推特.message.source
- 计数
-
要返回的最大事件数。默认值为 20。最多 50 人（整数，默认：
20)
5.19. Twitter 搜索源
Twitter 的标准搜索 API（搜索/推文）允许对最近或流行的推文的索引进行简单查询。这Source提供针对过去 7 天内发布的最近推文样本的连续搜索。“公共”API 集的一部分。
返回与指定查询匹配的相关推文的集合。
使用spring.cloud.stream.poller属性来控制连续搜索请求之间的间隔。速率限制 - 每 30 分钟窗口 180 个请求（例如 ~6 r/m，~ 1 req / 10 秒）
这twitter.search查询属性允许按关键字查询并按时间和地理位置过滤结果。
这twitter.search.count和twitter.search.page根据搜索 API 控制结果分页。
注意：Twitter 的搜索服务以及搜索 API 并不意味着成为推文的详尽来源。并非所有推文都会被索引或通过搜索界面提供。
5.19.1. 选项
按前缀分组的属性：
spring.integration.轮询器
- 克罗恩
-
用于轮询的 Cron 表达式。与 'fixedDelay' 和 'fixedRate' 互斥。（字符串，默认：
<none>) - 固定延迟
-
轮询延迟期。与 'cron' 和 'fixedRate' 互斥。（持续时间，默认：
<none>) - 固定利率
-
轮询率周期。与 'fixedDelay' 和 'cron' 互斥。（持续时间，默认：
<none>) - 初始延迟
-
轮询初始延迟。申请 'fixedDelay' 和 'fixedRate';“cron”被忽略。（持续时间，默认：
<none>) - 每次轮询的最大消息数
-
每个轮询周期要轮询的最大消息数。（整数，默认：
<none>) - 接收超时
-
轮询消息的等待时间。（持续时间，默认：
1s)
推特连接
- 访问Tokens
-
您的 Twitter Tokens。（字符串，默认：
<none>) - 访问Tokens密钥
-
你的 Twitter Tokens密钥。（字符串，默认：
<none>) - 消费者键
-
您的 Twitter 密钥。（字符串，默认：
<none>) - 消费者机密
-
你的 Twitter 秘密。（字符串，默认：
<none>) - 启用调试
-
启用 Twitter4J 调试模式。（布尔值，默认：
false) - 原始 JSON
-
启用缓存 Twitter API 返回的原始（原始）JSON 对象。当设置为 False 时，结果将使用 Twitter4J 的 json 表示形式。当设置为 True 时，结果将使用原始的 Twitter APISs json 表示形式。（布尔值，默认：
true)
推特.search
- 计数
-
每页要返回的推文数量（例如，每个请求），最多 100 条。（整数，默认：
100) - 朗
-
将搜索到的推文限制为给定语言，由 http://en.wikipedia.org/wiki/ISO_639-1 给出。（字符串，默认：
<none>) - 页
-
在再次从最近的推文开始搜索之前，要向后搜索（从最近的推文到最旧的推文）的页面数（例如请求）。向后搜索的推文总量为（页数*数）（整数，默认：
3) - 查询
-
按搜索查询字符串搜索推文。（字符串，默认：
<none>) - 从最近的空响应重新启动
-
从空回复的最新推文重新开始搜索。仅在第一次重启后应用（例如，当since_id ！= UNBOUNDED 时）（布尔值，默认：
false) - 结果类型
-
指定您希望接收的搜索结果类型。当前默认值为“混合”。有效值包括：混合 ：在响应中同时包含热门结果和实时结果。最近 ：仅返回响应中的最新结果 popular ：仅返回响应中最受欢迎的结果（ResultType，默认：
<none>，可能的值：popular,mixed,recent) - 因为
-
如果指定，则返回自给定日期以来的推文。日期的格式应为 YYYY-MM-DD。（字符串，默认：
<none>)
推特.search.geocode
- 纬度
-
用户的纬度。（双精度，默认：
-1) - 经度
-
用户的经度。（双精度，默认：
-1) - 半径
-
（纬度、经度）点周围的半径（以公里为单位）。（双精度，默认：
-1)
5.20. Twitter 流源
-
这
Filter API返回与一个或多个过滤器谓词匹配的公共状态。 多个参数允许使用与流式处理API的单个连接。 提示：这track,follow和locations字段与 OR 运算符组合！ 查询track=foo和follow=1234返回匹配的推文test或由用户创建1234. -
这
Sample API返回所有公共状态的小型随机样本。 默认访问级别返回的推文是相同的，因此，如果两个不同的客户端连接到此端点，它们将看到相同的推文。
默认访问级别允许多达 400 个跟踪关键字、5,000 个关注用户 ID 和 25 个 0.1-360 度位置框。
5.20.1. 选项
按前缀分组的属性：
推特连接
- 访问Tokens
-
您的 Twitter Tokens。（字符串，默认：
<none>) - 访问Tokens密钥
-
你的 Twitter Tokens密钥。（字符串，默认：
<none>) - 消费者键
-
您的 Twitter 密钥。（字符串，默认：
<none>) - 消费者机密
-
你的 Twitter 秘密。（字符串，默认：
<none>) - 启用调试
-
启用 Twitter4J 调试模式。（布尔值，默认：
false) - 原始 JSON
-
启用缓存 Twitter API 返回的原始（原始）JSON 对象。当设置为 False 时，结果将使用 Twitter4J 的 json 表示形式。当设置为 True 时，结果将使用原始的 Twitter APISs json 表示形式。（布尔值，默认：
true)
推特.stream.filter
- 计数
-
指示在过渡到实时流之前要流化的先前状态数。（整数，默认：
0) - 过滤器级别
-
筛选器级别将推文限制为具有最小 filterLevel 属性值的推文。无、低或中之一。（FilterLevel，默认：
<none>) - 跟随
-
按 ID 指定要从中接收公开推文的用户。（List<Long>，默认值：
<none>) - 语言
-
指定流的推文语言。（List<String>，默认值：
<none>) - 地点
-
要跟踪的位置。内部表示为 2D 数组。边界框无效：52.38、4.90、51.51、-0.12。第一对必须是盒子的SW角（List<BoundingBox>，默认值：
<none>) - 跟踪
-
指定要跟踪的关键字。（List<String>，默认值：
<none>)
推特.stream
- 类型
-
<缺少文档>（StreamType，默认：
<none>，可能的值：sample,filter,firehose,link)
5.21. Websocket 源代码
这Websocket通过 Web 套接字生成消息的源。
5.21.1. 选项
按前缀分组的属性：
websocket.supplier
- 允许的来源
-
允许的来源。（字符串，默认：
*) - 路径
-
服务器 WebSocket 处理程序公开的路径。（字符串，默认：
/websocket)
websocket.supplier.sock-js
- 使
-
在服务器上启用 SockJS 服务。默认值为“false”（布尔值，默认：
false)
5.21.2. 示例
要验证 websocket-source 是否从 Websocket 客户端接收消息，您可以使用以下简单的端到端设置。
第 1 步：启动 kafka
第 2 步：部署websocket-source在特定端口上，比如 8080
第 3 步：在端口 8080 路径“/websocket”上连接一个 websocket 客户端，并发送一些消息。
您可以启动 kafka 控制台使用者并在那里查看消息。
5.22. XMPP 源
“xmpp”源允许从 XMPP 服务器接收消息。
5.22.1. 输入
不适用
5.22.2. 输出
有效载荷
-
byte[]
5.22.3. 选项
xmpp 源具有以下选项：
按前缀分组的属性：
xmpp.factory
- 主机
-
要连接到的 XMPP 主机服务器。（字符串，默认：
<none>) - 密码
-
连接用户的密码。（字符串，默认：
<none>) - 端口
-
用于连接到主机的端口。- 默认客户端端口：5222（整数，默认：
5222) - 资源
-
要绑定到 XMPP 主机上的资源。- 可以为空，如果未设置，服务器将生成一个（字符串，默认：
<none>) - 安全模式
-
<缺少文档>（SecurityMode，默认：
<none>，可能的值：required,ifpossible,disabled) - 服务名称
-
要为 XMPP 域设置的服务名称。（字符串，默认：
<none>) - 订阅模式
-
<缺少文档>（SubscriptionMode，默认：
<none>，可能的值：accept_all,reject_all,manual) - 用户
-
连接应作为的用户。（字符串，默认：
<none>)
xmpp.提供商
- 有效负载表达式
-
<缺少文档>（表达式，默认值：
<none>) - 节过滤器
-
<缺少文档>（StanzaFilter，默认：
<none>)
另请参阅 Spring Boot 文档，了解代理连接和侦听器属性的附加属性。
5.22.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps您可以在此处找到相应的基于活页夹的项目。 然后，您可以 cd 到其中一个文件夹中并构建它：
$ ./mvnw clean package5.22.5. 示例
java -jar xmpp-source.jar --xmpp.factory.host=localhost --xmpp.factory.port=5222 --xmpp.factory.user=jane --xmpp.factory.password=secret --xmpp.factory.service-name=localhost5.23. ZeroMQ 源代码
“zeromq”源允许从 ZeroMQ 接收消息。
5.23.1. 输入
不适用
5.23.2. 输出
有效载荷
-
byte[]
5.23.3. 选项
zeromq 源具有以下选项：
- zeromq.supplier.bind-port
-
用于创建 ZeroMQ 套接字的绑定端口;0 选择一个随机端口。（整数，默认：
0) - zeromq.supplier.connect-url
-
到 ZeroMQ 套接字的连接 URL。（字符串，默认：
<none>) - zeromq.supplier.消费延迟
-
未收到数据时从 ZeroMQ 套接字消费的延迟。（持续时间，默认：
1s) - zeromq.supplier.socket-type
-
连接应建立的套接字类型。（SocketType，默认：
<none>，可能的值：PAIR,PUB,SUB,REQ,REP,DEALER,ROUTER,PULL,PUSH,XPUB,XSUB,STREAM,CLIENT,SERVER,RADIO,DISH,CHANNEL,PEER,RAW,SCATTER,GATHER) - zeromq.supplier.topics
-
要订阅的主题。（String[]，默认：
[])
另请参阅 Spring Boot 文档，了解代理连接和侦听器属性的附加属性。
5.23.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps您可以在此处找到相应的基于活页夹的项目。 然后，您可以 cd 到其中一个文件夹中并构建它：
$ ./mvnw clean package5.23.5. 示例
java -jar zeromq-source.jar --zeromq.supplier.connectUrl=tcp://server:port --zeromq.supplier.topics=6. 处理器
6.1. 聚合处理器
聚合器处理器使应用程序能够将传入消息聚合到组中，并将它们释放到输出目标中。
java -jar aggregator-processor-kafka-<version>.jar --aggregator.message-store-type=jdbc
如果要针对 RabbitMQ 运行 kafka，请将其更改为 rabbit。
6.1.1. 有效载荷
如果输入有效负载是byte[]并且 content-type 标头是 JSON，则JsonBytesToMap函数尝试将此有效负载反序列化为Map以便在聚合器函数的输出上更好地表示数据。
此外，这样的Map数据表示使从下面提到的 SpEL 表达式中轻松访问有效负载内容变得容易。
否则（包括反序列化错误），输入有效负载将保持原样 - 它是目标应用程序配置，用于将其转换为所需的形式。
6.1.2. 选项
按前缀分组的属性：
聚合
- 集合体
-
用于聚集策略的 SpEL 表达。默认值为有效负载的集合。（表达式，默认值：
<none>) - 相关
-
相关键的 SpEL 表达式。默认为 correlationId 标头。（表达式，默认值：
<none>) - 组超时
-
SpEL 表达式，用于超时到即将过期的未完成组。（表达式，默认值：
<none>) - 消息存储实体
-
持久性消息存储实体：RDBMS 中的表前缀、MongoDb 中的集合名称等。（字符串，默认：
<none>) - 消息存储类型
-
消息存储类型。（字符串，默认：
<none>) - 释放
-
用于释放策略的 SpEL 表达式。默认值基于 sequenceSize 标头。（表达式，默认值：
<none>)
spring.data.mongodb
- 附加主机
-
其他服务器主机。不能使用 URI 设置，或者如果未指定“host”。其他主机将使用默认的 mongo 端口 27017。如果要使用不同的端口，可以使用“host：port”语法。（List<String>，默认值：
<none>) - 身份验证数据库
-
身份验证数据库名称。（字符串，默认：
<none>) - 自动索引创建
-
是否启用自动索引创建。（布尔值，默认：
<none>) - 数据库
-
数据库名称。覆盖 URI 中的数据库。（字符串，默认：
<none>) - 字段命名策略
-
要使用的 FieldNamingStrategy 的完全限定名称。（Class<？>，默认：
<none>) - 主机
-
Mongo 服务器主机。不能使用 URI 设置。（字符串，默认：
<none>) - 密码
-
mongo 服务器的登录密码。不能使用 URI 设置。（Character[]，默认：
<none>) - 端口
-
Mongo 服务器端口。不能使用 URI 设置。（整数，默认：
<none>) - 副本集名称
-
集群所需的副本集名称。不能使用 URI 设置。（字符串，默认：
<none>) - URI
-
Mongo 数据库 URI。覆盖主机、端口、用户名和密码。（字符串，默认：
mongodb://localhost/test) - 用户名
-
mongo 服务器的登录用户。不能使用 URI 设置。（字符串，默认：
<none>) - uuid-表示
-
将 UUID 转换为 BSON 二进制值时使用的表示。（UuidRepresentation，默认：
java-legacy，可能的值：UNSPECIFIED,STANDARD,C_SHARP_LEGACY,JAVA_LEGACY,PYTHON_LEGACY)
spring.data.mongodb.gridfs
- 桶
-
GridFS 存储桶名称。（字符串，默认：
<none>) - 数据库
-
GridFS 数据库名称。（字符串，默认：
<none>)
spring.data.mongodb.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<none>) - 启用
-
是否启用 SSL 支持。除非另有说明，否则如果提供“捆绑包”，则自动启用。（布尔值，默认：
<none>)
spring.data.redis
- 客户端名称
-
要在与 CLIENT SETNAME 的连接上设置的客户端名称。（字符串，默认：
<none>) - 客户端类型
-
要使用的客户端类型。默认情况下，根据类路径自动检测。（ClientType，默认：
<none>，可能的值：LETTUCE,JEDIS) - 连接超时
-
连接超时。（持续时间，默认：
<none>) - 数据库
-
连接工厂使用的数据库索引。（整数，默认：
0) - 主机
-
Redis 服务器主机。（字符串，默认：
localhost) - 密码
-
redis 服务器的登录密码。（字符串，默认：
<none>) - 端口
-
Redis 服务器端口。（整数，默认：
6379) - 超时
-
读取超时。（持续时间，默认：
<none>) - 网址
-
连接 URL。覆盖主机、端口、用户名和密码。示例：redis://user：[email protected]：6379（字符串，默认：
<none>) - 用户名
-
redis 服务器的登录用户名。（字符串，默认：
<none>)
spring.data.redis.cluster
- 最大重定向
-
跨集群执行命令时要遵循的最大重定向数。（整数，默认：
<none>) - 节点
-
要从中引导的“host：port”对的逗号分隔列表。这表示集群节点的“初始”列表，并且需要至少有一个条目。（List<String>，默认值：
<none>)
spring.data.redis.jedis.pool
- 启用
-
是否启用池。如果“commons-pool2”可用，则自动启用。使用 Jedis，在哨兵模式下隐式启用池化，并且此设置仅适用于单节点设置。（布尔值，默认：
<none>) - 最大活动
-
池在给定时间可以分配的最大连接数。使用负值表示无限制。（整数，默认：
8) - 最大空闲
-
池中“空闲”连接的最大数量。使用负值表示无限数量的空闲连接。（整数，默认：
8) - 最大等待
-
在池耗尽时引发异常之前，连接分配应阻止的最长时间。使用负值无限期阻止。（持续时间，默认：
-1ms) - 最小空闲
-
池中要维护的最小空闲连接数的目标。仅当此设置和逐出运行之间的时间均为正时，此设置才有效。（整数，默认：
0) - 驱逐运行之间的时间
-
空闲对象 evictor 线程运行之间的时间。当为正时，空闲对象 evictor 线程将启动，否则不会执行空闲对象逐出。（持续时间，默认：
<none>)
spring.data.redis.lettuce.pool
- 启用
-
是否启用池。如果“commons-pool2”可用，则自动启用。使用 Jedis，在哨兵模式下隐式启用池化，并且此设置仅适用于单节点设置。（布尔值，默认：
<none>) - 最大活动
-
池在给定时间可以分配的最大连接数。使用负值表示无限制。（整数，默认：
8) - 最大空闲
-
池中“空闲”连接的最大数量。使用负值表示无限数量的空闲连接。（整数，默认：
8) - 最大等待
-
在池耗尽时引发异常之前，连接分配应阻止的最长时间。使用负值无限期阻止。（持续时间，默认：
-1ms) - 最小空闲
-
池中要维护的最小空闲连接数的目标。仅当此设置和逐出运行之间的时间均为正时，此设置才有效。（整数，默认：
0) - 驱逐运行之间的时间
-
空闲对象 evictor 线程运行之间的时间。当为正时，空闲对象 evictor 线程将启动，否则不会执行空闲对象逐出。（持续时间，默认：
<none>)
spring.data.redis.lettuce
- 关机超时
-
关机超时。（持续时间，默认：
100ms)
spring.data.redis.sentinel
- 主人
-
Redis 服务器的名称。（字符串，默认：
<none>) - 节点
-
逗号分隔的“host：port”对列表。（List<String>，默认值：
<none>) - 密码
-
用于向哨兵进行身份验证的密码。（字符串，默认：
<none>) - 用户名
-
用于向哨兵进行身份验证的登录用户名。（字符串，默认：
<none>)
spring.data.redis.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<none>) - 启用
-
是否启用 SSL 支持。除非另有说明，否则如果提供“捆绑包”，则自动启用。（布尔值，默认：
<none>)
spring.datasource
- 驱动程序类名称
-
JDBC 驱动程序的完全限定名称。默认情况下，根据 URL 自动检测。（字符串，默认：
<none>) - 密码
-
数据库的登录密码。（字符串，默认：
<none>) - 网址
-
数据库的 JDBC URL。（字符串，默认：
<none>) - 用户名
-
数据库的登录用户名。（字符串，默认：
<none>)
6.3. 滤波处理器
筛选处理器使应用程序能够检查传入的有效负载，然后对其应用谓词，以决定是否需要继续记录。
例如，如果传入有效负载的类型为String如果您想过滤掉任何少于五个字符的内容，您可以运行过滤处理器，如下所示。
java -jar filter-processor-kafka-<version>.jar --filter.function.expression=payload.length() > 4
如果要针对 RabbitMQ 运行 kafka，请将其更改为 rabbit。
6.3.1. 有效载荷
您可以将任何类型作为有效负载传递，然后对其应用 SpEL 表达式进行过滤。
如果传入类型为byte[]并将内容类型设置为text/plain或application/json，则应用程序将byte[]到String.
6.3.2. 选项
- filter.function.expression
-
布尔 SpEL 表达式，用于应用于要过滤的请求消息。（表达式，默认值：
<none>)
6.4. Groovy 处理器
对消息应用 Groovy 脚本的处理器。
6.4.1. 选项
groovy-processor 处理器具有以下选项：
- groovy-processor.script
-
引用用于处理消息的脚本。（资源，默认值：
<none>) - groovy-processor.变量
-
变量绑定为换行符，以名称-值对为分隔字符串，例如 'foo=bar\n baz=car'。（属性，默认值：
<none>) - groovy-processor.变量位置
-
包含自定义脚本变量绑定的属性文件的位置。（资源，默认值：
<none>)
6.5. 标头富集处理器
使用标头扩充器应用添加邮件标头。
标头以换行符分隔键值对的形式提供，其中键是标头名称，值是 SpEL 表达式。
例如--headers='foo=payload.someProperty \n bar=payload.otherProperty'.
6.5.1. 选项
标头扩充处理器具有以下选项：
- header.enricher.headers
-
\n 表示标头的分隔属性，其中值是 SpEL 表达式，例如 foo='bar' \n baz=payload.baz。（属性，默认值：
<none>) - header.enricher.overwrite
-
设置为 true 以覆盖任何现有邮件标头。（布尔值，默认：
false)
6.6. Http 请求处理器
向 HTTP 资源发出请求并将响应正文作为消息有效负载发出的处理器应用程序。
6.6.1. 输入
头
任何必需的 HTTP 标头都必须通过headers或headers-expression财产。请参阅下面的示例。
标头值也可用于构造：
-
请求正文在
body-expression财产。 -
HTTP 方法在
http-method-expression财产。 -
在
url-expression财产。
有效载荷
默认情况下，有效负载用作 POST 请求的请求正文，并且可以是任何 Java 类型。对于 GET 请求，它应该是一个空字符串。有效负载也可用于构造：
-
请求正文在
body-expression财产。 -
HTTP 方法在
http-method-expression财产。 -
在
url-expression财产。
底层 WebClient 支持 Jackson JSON 序列化，以支持任何请求和响应类型（如有必要）。 这expected-response-type属性String.class默认情况下，可以设置为应用程序类路径中的任何类。请注意，用户定义的有效负载类型需要向 pom 文件添加所需的依赖项。
6.6.2. 输出
头
没有 HTTP 消息头映射到出站消息。
有效载荷
原始输出对象是 ResponseEntity<？>其任何字段（例如body,headers） 或访问器方法 （statusCode） 可以作为reply-expression.
默认情况下，出站消息有效负载是响应正文。
请注意，ResponseEntity（由表达式#root） 默认情况下不能被 Jackson 反序列化，但可以呈现为HashMap.
6.6.3. 选项
http-request 处理器具有以下选项：
6.6.4. 选项
按前缀分组的属性：
http.请求
- 身体表达
-
一个 SpEL 表达式，用于从传入消息派生请求正文。（表达式，默认值：
<none>) - 预期响应类型
-
用于解释响应的类型。（Class<？>，默认：
<none>) - 标头表达式
-
用于派生要使用的 http 标头映射的 SpEL 表达式。（表达式，默认值：
<none>) - http 方法表达式
-
一个 SpEL 表达式，用于从传入消息派生请求方法。（表达式，默认值：
<none>) - 回复表达式
-
用于计算最终结果的 SpEL 表达式，应用于整个 http {@link org.springframework.http.ResponseEntity}。（表达式，默认值：
<none>) - 超时
-
请求超时（以毫秒为单位）。（长，默认：
30000) - url 表达式
-
针对传入消息的 SpEL 表达式，用于确定要使用的 URL。（表达式，默认值：
<none>)
spring.编解码器
- 日志请求详细信息
-
是否在 DEBUG 级别记录表单数据，并在 TRACE 级别记录标头。（布尔值，默认：
false) - 内存中最大大小
-
每当需要聚合输入流时可以缓冲的字节数限制。这仅适用于自动配置的 WebFlux 服务器和 WebClient 实例。默认情况下，这不会设置，在这种情况下，单个编解码器默认值适用。默认情况下，大多数编解码器限制为 256K。（DataSize，默认：
<none>)
6.7. 图像识别处理器
使用 Inception 模型对实时图像进行分类的处理器（例如标签）。
模型的输入是作为二进制数组的图像。
输出是以下格式的 JSON 消息：
{
"labels" : [
{"giant panda":0.98649305}
]
}结果包含已识别类别的名称（例如标签）以及图像代表此类别的置信度（例如置信度）。
如果response-seize设置为高于 1 的值，则结果将包括顶部response-seize可能的标签。例如response-size=3将返回：
{
"labels": [
{"giant panda":0.98649305},
{"badger":0.010562794},
{"ice bear":0.001130851}
]
}6.7.1. 有效载荷
如果传入类型为byte[]并将内容类型设置为application/octet-stream，则应用程序处理输入byte[]图像输入并输出增强byte[]图像有效负载和 JSON 标头。
6.7.2. 选项
- image.recognition.cache-model
-
缓存预训练的 TensorFlow 模型。（布尔值，默认：
true) - image.recognition.debug-output
-
<缺少文档>（布尔值，默认：
false) - image.recognition.debug-output-path
-
<缺少文档>（字符串，默认：
image-recognition-result.png) - image.recognition.model（图像识别模型）
-
预训练的 TensorFlow 图像识别模型。请注意，模型必须与所选模型类型匹配！（字符串，默认：
https://storage.googleapis.com/mobilenet_v2/checkpoints/mobilenet_v2_1.4_224.tgz#mobilenet_v2_1.4_224_frozen.pb) - image.recognition.model-type
-
支持三种不同的预训练 tensorflow 图像识别模型：Inception、MobileNetV1 和 MobileNetV2 1。Inception 图使用“输入”作为输入，使用“输出”作为输出。2. MobileNetV2 预训练模型：https://github.com/tensorflow/models/tree/master/research/slim/nets/mobilenet#pretrained-models - 归一化图像大小始终为正方形（例如 H=W） - 图形使用“input”作为输入，使用“MobilenetV2/Predictions/Reshape_1”作为输出。3. MobileNetV1 预训练模型：https://github.com/tensorflow/models/blob/master/research/slim/nets/mobilenet_v1.md#pre-trained-models - graph 使用“input”作为输入，“MobilenetV1/Predictions/Reshape_1”作为输出。（ModelType，默认：
<none>，可能的值：inception,mobilenetv1,mobilenetv2) - image.recognition.normalized-image-size
-
标准化图像大小。（整数，默认：
224) - image.recognition.response-size
-
识别的图像数量。（整数，默认：
5)
6.8. 对象检测处理器
对象检测处理器为 TensorFlow 对象检测 API 提供开箱即用的支持。它允许实时定位和识别单个图像或图像流中的多个对象。对象检测处理器建立在对象检测功能之上。
以下是一些合理的配置默认值：
-
object.detection.model:storage.googleapis.com/scdf-tensorflow-models/object-detection/faster_rcnn_resnet101_coco_2018_01_28_frozen_inference_graph.pb -
object.detection.labels:storage.googleapis.com/scdf-tensorflow-models/object-detection/mscoco_label_map.pbtxt -
object.detection.with-masks:false
下图显示了一个 Spring Cloud Data Flow，流管道，它实时预测输入图像流中的对象类型。
处理器的输入是一个图像字节数组，输出是一个增强图像和一个标头，称为detected_objects，提供检测到的对象的文本描述：
{
"labels" : [
{"name":"person", "confidence":0.9996774,"x1":0.0,"y1":0.3940161,"x2":0.9465165,"y2":0.5592592,"cid":1},
{"name":"person", "confidence":0.9996604,"x1":0.047891676,"y1":0.03169123,"x2":0.941098,"y2":0.2085562,"cid":1},
{"name":"backpack", "confidence":0.96534747,"x1":0.15588468,"y1":0.85957795,"x2":0.5091308,"y2":0.9908878,"cid":23},
{"name":"backpack", "confidence":0.963343,"x1":0.1273736,"y1":0.57658505,"x2":0.47765,"y2":0.6986431,"cid":23}
]
}这detected_objects标头格式为：
-
object-name：confidence - 检测到的对象（例如标签）的人类可读名称，其置信度为 [0-1] 之间的浮点数
-
x1， y1， x2， y2 - 响应还提供检测到的对象的边界框，表示为
(x1, y1, x2, y2).坐标是相对于图像大小的。 -
cid - 提供的标签配置文件中定义的分类标识符。
6.8.1. 有效载荷
传入类型为byte[]，内容类型为application/octet-stream.处理器处理输入byte[]图像并输出增强的byte[]图像有效负载和 JSON 标头 （detected_objects).
6.8.2. 选项
- 对象.检测.缓存模型
-
<缺少文档>（布尔值，默认：
true) - 对象.检测.置信度
-
<缺少文档>（浮点，默认：
0.4) - object.detection.debug-output
-
<缺少文档>（布尔值，默认：
false) - 对象.检测.调试输出路径
-
<缺少文档>（字符串，默认：
object-detection-result.png) - 对象.检测.标签
-
标记 URI。（字符串，默认：
https://storage.googleapis.com/scdf-tensorflow-models/object-detection/mscoco_label_map.pbtxt) - 对象.检测.模型
-
预训练的 TensorFlow 对象检测模型。（字符串，默认：
https://download.tensorflow.org/models/object_detection/ssdlite_mobilenet_v2_coco_2018_05_09.tar.gz#frozen_inference_graph.pb) - 对象.检测.响应大小
-
<缺少文档>（整数，默认：
<none>) - 对象.detection.with-masks
-
<缺少文档>（布尔值，默认：
false)
6.9. 语义分割处理器
基于最先进的 DeepLab Tensorflow 模型的图像语义分割。
这Semantic Segmentation是将图像的每个像素与类标签（例如花、人、道路、天空、海洋或汽车）相关联的过程。
与Instance Segmentation，它会生成实例感知区域掩码，则Semantic Segmentation生成类感知掩码。
用于实施Instance Segmentation请改为咨询对象检测服务。
这Semantic Segmentation Processor使用语义分割函数库和 TensorFlow 服务。
6.9.1. 有效载荷
传入类型为byte[]，内容类型为application/octet-stream.处理器处理输入byte[]图像和输出增强byte[]图像有效负载和 JSON 标头。
处理器的输入是一个图像字节数组，输出是一个增强的图像字节数组和一个 JSON 标头semantic_segmentation在这种格式中：
[
[ 0, 0, 0 ],
[ 127, 127, 127 ],
[ 255, 255, 255 ],
...
]输出标头 json 格式表示从输入图像计算的颜色像素贴图。
6.9.2. 选项
- semantic.segmentation.color-map-uri
-
每个预训练模型都基于某些对象颜色图。预定义的选项是： - classpath：/colormap/citymap_colormap.json - classpath：/colormap/ade20k_colormap.json - classpath：/colormap/black_white_colormap.json - classpath：/colormap/mapillary_colormap.json（字符串，默认：
classpath:/colormap/citymap_colormap.json) - semantic.segmentation.debug-output
-
保存本地 debugOutputPath 路径中的输出映像。（布尔值，默认：
false) - semantic.segmentation.debug-output-path
-
<缺少文档>（字符串，默认：
semantic-segmentation-result.png) - semantic.segmentation.mask-transparency
-
计算的分割掩码图像的 alpha 颜色。（浮点，默认：
0.45) - 语义.segmentation.model
-
预训练的 TensorFlow 语义分割模型。（字符串，默认：
https://download.tensorflow.org/models/deeplabv3_mnv2_cityscapes_train_2018_02_05.tar.gz#frozen_inference_graph.pb) - semantic.segmentation.output-type
-
指定输出图像类型。您可以返回带有计算掩码叠加的输入图像，也可以单独返回掩码。（OutputType，默认：
<none>，可能的值：blended,mask)
6.10. 脚本处理器
使用脚本转换消息的处理器。脚本主体直接提供作为属性值。可以指定脚本的语言（groovy/javascript/ruby/python）。
6.10.1. 选项
脚本处理器处理器具有以下选项：
- 脚本处理器.language
-
script 属性中文本的语言。支持：groovy、javascript、ruby、python。（字符串，默认：
<none>) - 脚本处理器.script
-
脚本文本。（字符串，默认：
<none>) - 脚本处理器.变量
-
变量绑定为换行符，以名称-值对为分隔字符串，例如 'foo=bar\n baz=car'。（属性，默认值：
<none>) - 脚本处理器.变量位置
-
包含自定义脚本变量绑定的属性文件的位置。（资源，默认值：
<none>)
6.11. 拆分器处理器
拆分器应用程序建立在 Spring Integration 中同名的概念之上，并允许将单个消息拆分为多个不同的消息。
处理器使用一个函数，该函数接受Message<?>作为输入，然后生成一个List<Message<?>作为基于各种属性的输出（见下文）。
您可以使用 SpEL 表达式或分隔符来指定要如何拆分传入消息。
6.11.1. 有效载荷
-
传入有效载荷 -
Message<?>
如果传入类型为byte[]并将内容类型设置为text/plain或application/json，则应用程序将byte[]到String.
-
传出有效载荷 -
List<Message<?>
6.11.2. 选项
- splitter.apply-sequence
-
在标头中添加相关性/序列信息，以方便以后聚合。（布尔值，默认：
true) - splitter.charset
-
将基于文本的文件中的字节转换为 String 时要使用的字符集。（字符串，默认：
<none>) - splitter.delimiters
-
当表达式为 null 时，标记 {@link String} 有效负载时要使用的分隔符。（字符串，默认：
<none>) - splitter.expression
-
用于拆分有效负载的 SpEL 表达式。（字符串，默认：
<none>) - splitter.文件标记
-
设置为 true 或 false 以使用包含（或不包含）文件开头/结尾标记的 {@code FileSplitter}（按行拆分基于文本的文件）。（布尔值，默认：
<none>) - splitter.markers-json
-
当 'fileMarkers == true' 时，指定它们是否应生成为 FileSplitter.FileMarker 对象或 JSON。（布尔值，默认：
true)
6.12. 变换处理器
Transformer 处理器允许您根据 SpEL 表达式转换消息有效负载结构。
下面是如何运行此应用程序的示例。
java -jar transform-processor-kafka-<version>.jar \
--spel.function.expression=payload.toUpperCase()如果要针对 RabbitMQ 运行 kafka，请将其更改为 rabbit。
6.12.1. 有效载荷
传入消息可以包含任何类型的有效负载。
6.12.2. 选项
- spel.function.expression
-
要应用的 SpEL 表达式。（字符串，默认：
<none>)
6.13. Twitter 趋势和趋势位置处理器
可以返回趋势主题或趋势主题的位置的处理器。
这twitter.trend.trend-query-type属性允许来选择查询类型。
6.13.1. 检索某个位置的趋势主题（可选）
对于此模式集twitter.trend.trend-query-type自trend.
6.13.2. 检索趋势位置
对于此模式集twitter.trend.trend-query-type自trendLocation.
按位置检索热门主题的完整或附近位置列表。
如果latitude,longitude参数不提供处理器执行可用趋势 API 并返回 Twitter 具有趋势主题信息的位置。
如果latitude,longitude参数，则处理器执行 Trends Closest API 并返回 Twitter 具有最接近指定位置的趋势主题信息的位置。
Response 是locations对该位置的 WOEID 和其他一些人类可读信息（例如该位置所属的规范名称和国家/地区）进行编码。
6.13.3. 选项
按前缀分组的属性：
推特.trend.closest
- 纬度
-
如果提供了长参数，则可用的趋势位置将按距离（从最近到最远）到坐标对进行排序。经度的有效范围为 -180.0 到 +180.0（西为负，东为正）（含）。（表达式，默认值：
<none>) - 离子
-
如果提供了纬度参数，则可用的趋势位置将按距离（从最近到最远）到坐标对进行排序。经度的有效范围为 -180.0 到 +180.0（西为负，东为正）（含）。（表达式，默认值：
<none>)
推特.trend
- 位置 ID
-
雅虎！要返回趋势信息的位置的地球上 ID。使用 1 作为 WOEID 即可获得全局信息。（表达式，默认值：
payload) - 趋势查询类型
-
<缺少文档>（TrendQueryType，默认：
<none>，可能的值：trend,trendLocation)
7. 水槽
7.1. 卡桑德拉水槽
此接收器应用程序将收到的每条消息的内容写入 Cassandra。
它需要 JSON 字符串的有效负载，并使用其属性映射到表列。
7.1.1. 有效载荷
表示要持久化的实体（或实体列表）的 JSON 字符串或字节数组。
7.1.2. 选项
cassandra 水槽具有以下选项：
按前缀分组的属性：
cassandra.集群
- 创建键空间
-
在应用程序启动时创建（或不创建）键空间的标志。（布尔值，默认：
false) - 实体基础包
-
用于扫描带有 Table 注释的实体的基本包。（String[]，默认：
[]) - 初始化脚本
-
具有 CQL 脚本（以 ';' 分隔）的资源，用于初始化密钥空间模式。（资源，默认值：
<none>) - 跳过 SSL 验证
-
用于验证服务器的 SSL 证书的标志。（布尔值，默认：
false)
Cassandra
- 一致性级别
-
写入作的一致性级别。（ConsistencyLevel，默认：
<none>) - 摄取查询
-
引入 Cassandra 查询。（字符串，默认：
<none>) - 查询类型
-
Cassandra 接收器的 QueryType。（类型，默认值：
<none>，可能的值：INSERT,UPDATE,DELETE,STATEMENT) - 语句表达式
-
Cassandra 查询 DSL 样式中的表达式。（表达式，默认值：
<none>) - TTL的
-
Time-to-live 选项。（整数，默认：
0)
spring.cassandra
- 压缩
-
Cassandra 二进制协议支持压缩。（压缩，默认：
none，可能的值：LZ4,SNAPPY,NONE) - 配置
-
要使用的配置文件的位置。（资源，默认值：
<none>) - 接触点
-
集群节点地址格式为“host：port”，或简单的“host”，以使用配置的端口。（List<String>，默认值：
[127.0.0.1:9042]) - 键空间名称
-
要使用的键空间名称。（字符串，默认：
<none>) - 本地数据中心
-
被视为“本地”的数据中心。联系点应来自此数据中心。（字符串，默认：
<none>) - 密码
-
服务器的登录密码。（字符串，默认：
<none>) - 端口
-
如果触点未指定一个端口，则要使用的端口。（整数，默认：
9042) - 模式作
-
启动时要执行的架构作。（字符串，默认：
none) - 会话名称
-
Cassandra 会话的名称。（字符串，默认：
<none>) - 用户名
-
服务器的登录用户。（字符串，默认：
<none>)
spring.cassandra.连接
- 连接超时
-
建立驱动程序连接时要使用的超时。（持续时间，默认：
5s) - 初始查询超时
-
超时，用于在初始化过程中作为初始化过程的一部分运行的内部查询，就在打开连接后。（持续时间，默认：
5s)
spring.cassandra.control连接
- 超时
-
用于控制查询的超时。（持续时间，默认：
5s)
spring.cassandra.pool
- 心跳间隔
-
在空闲连接上发送消息以确保其仍然处于活动状态的检测信号间隔。（持续时间，默认：
30s) - 空闲超时
-
删除空闲连接之前的空闲超时。（持续时间，默认：
5s)
spring.cassandra.请求
- 一致性
-
查询一致性级别。（DefaultConsistencyLevel，默认：
<none>，可能的值：ANY,ONE,TWO,THREE,QUORUM,ALL,LOCAL_ONE,LOCAL_QUORUM,EACH_QUORUM,SERIAL,LOCAL_SERIAL) - 页面大小
-
在单个网络往返中同时检索多少行。（整数，默认：
5000) - 串行一致性
-
查询串行一致性级别。（DefaultConsistencyLevel，默认：
<none>，可能的值：ANY,ONE,TWO,THREE,QUORUM,ALL,LOCAL_ONE,LOCAL_QUORUM,EACH_QUORUM,SERIAL,LOCAL_SERIAL) - 超时
-
驱动程序等待请求完成的时间。（持续时间，默认：
2s)
spring.cassandra.request.throttler
- 排水间隔
-
限制器尝试将请求出队的频率。将此设置设置得足够高，以便每次尝试都会处理队列中的多个条目，但不会将请求延迟太多。（持续时间，默认：
<none>) - 最大并发请求数
-
允许并行执行的最大请求数。（整数，默认：
<none>) - 最大队列大小
-
超过限制阈值时可以排队的最大请求数。（整数，默认：
<none>) - 每秒最大请求数
-
允许的最大请求速率。（整数，默认：
<none>) - 类型
-
请求限制类型。（ThrottlerType，默认：
none，可能的值：CONCURRENCY_LIMITING,RATE_LIMITING,NONE)
spring.cassandra.ssl 的
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<none>) - 启用
-
是否启用 SSL 支持。（布尔值，默认：
<none>)
7.2. 分析接收器
Sink 应用程序，构建在 Analytics 使用者之上，它根据输入消息计算分析，并将分析作为指标发布到各种监控系统。它利用千分尺库在最流行的监控系统中提供统一的编程体验，并公开了 Spring 表达式语言 （SpEL） 属性，用于定义如何从输入数据计算指标名称、值和标签。
分析接收器可以生成两种指标类型：
仪表（例如计数器或仪表）由其唯一标识name和dimensions（术语“尺寸”和“标记”可以互换使用）。维度允许对特定命名指标进行切片，以向下钻取和推理数据。
作为指标，由其name和dimensions，您可以为每个指标分配多个标签（e.g. key/值对），但之后不能随机更改这些标签！Prometheus 等监控系统会抱怨，如果具有相同名称的指标具有不同的标签集。 |
使用analytics.name或analytics.name-expression属性 设置输出分析指标的名称。如果未设置，则指标名称默认为应用程序名称。
使用analytics.tag.expression.<TAG_NAME>=<TAG_VALUE>，属性，用于向指标添加一个或多个标记。这TAG_NAME属性定义中使用的将在指标中显示为标记名称。TAG_VALUE是一个SpEL从传入消息动态计算标记值的表达式。
这SpEL表达式使用headers和payload关键字来访问邮件的标头和有效负载值。
您可以使用文字（例如'fixed value'） 设置具有固定值的标签。 |
所有 Stream 应用程序都支持，盒子的 ouf，三个最流行的监控系统，Wavefront,Prometheus和InfluxDB并且您可以以声明方式启用它们中的每一个。
只需将 micrometer meter-registry 依赖项添加到Analytics Sink应用。
请访问 Spring Cloud Data Flow Stream Monitoring 以获取配置监控系统的详细说明。以下快速片段可以帮助您入门。
-
要启用 Prometheus 仪表注册表，请设置以下属性。
management.metrics.export.prometheus.enabled=true
management.metrics.export.prometheus.rsocket.enabled=true
management.metrics.export.prometheus.rsocket.host=<YOUR PROMETHEUS-RSOKET PROXI URI
management.metrics.export.prometheus.rsocket.port=7001-
要启用 Wavefront 仪表注册表，请设置以下属性。
management.metrics.export.wavefront.enabled=true
management.metrics.export.wavefront.api-token=YOUR WAVEFRONT KEY
management.metrics.export.wavefront.uri=YOUR WAVEFRONT URI
management.metrics.export.wavefront.source=UNIQUE NAME TO IDENTIFY YOUR APP-
要启用 InfluxDB 仪表注册表，请设置以下属性。
management.metrics.export.influx.enabled=true
management.metrics.export.influx.uri={influxdb-server-url}
如果启用了数据流服务器监控，则Analytics Sink将重用提供的指标配置。 |
下图说明了如何Analytics Sink帮助收集企业内部信息，用于证券交易所，实时管道。
7.2.1. 有效载荷
传入消息可以包含任何类型的有效负载。
7.2.2. 选项
按前缀分组的属性：
分析学
- 金额表达式
-
用于计算输出指标值（例如金额）的 SpEL 表达式。它默认为 1.0（表达式，默认值：
<none>) - 仪表型
-
千分尺计类型，用于向后端报告指标。（MeterType，默认：
<none>，可能的值：counter,gauge) - 名称
-
输出指标的名称。'name' 和 'nameExpression' 是互斥的。只能设置其中一个。（字符串，默认：
<none>) - 名称表达式
-
一个 SpEL 表达式，用于从输入消息中计算输出指标名称。'name' 和 'nameExpression' 是互斥的。只能设置其中一个。（表达式，默认值：
<none>)
analytics.tag
- 表达
-
根据 SpEL 表达式计算标记。单个 SpEL 表达式可以生成值数组，这反过来又意味着不同的名称/值标记。每个名称/值标记都会产生单独的计量增量。标记表达式格式为：analytics.tag.expression。[标记名称]=[SpEL 表达式]（Map<String， Expression>，默认值：
<none>) - 固定
-
已弃用：请将 analytics.tag.expression 与文字 SpEL 表达式一起使用。自定义、固定标签。这些标记具有常量值，创建一次，然后与每个已发布的指标一起发送。定义固定标记的约定为：<code>analytics.tag.fixed。[标签名称]=[标签值] </code>（Map<String， String>，默认：
<none>)
7.3. Elasticsearch Sink
将文档索引到 Elasticsearch 中的 Sink。
此 Elasticsearch 接收器仅支持索引 JSON 文档。
它使用来自输入目标的数据，然后将其索引到 Elasticsearch。
输入数据可以是纯 json 字符串，也可以是java.util.Map表示 JSON。
它还接受 Elasticsearch 提供的数据XContentBuilder.
但是，这是一种罕见的情况，因为中间件不太可能将记录保留为XContentBuilder.
这主要用于直接调用消费者。
7.3.1. 选项
Elasticsearch 接收器具有以下选项：
按前缀分组的属性：
弹性搜索.consumer
- 异步
-
指示索引作是否异步。默认情况下，索引是同步完成的。（布尔值，默认：
false) - 批量大小
-
每个请求要索引的项数。默认为 1。对于大于 1 的值，将使用批量索引 API。（整数，默认：
1) - 组超时
-
超时（以毫秒为单位），之后在批量索引处于活动状态时刷新消息组。它默认为 -1，这意味着不会自动刷新空闲消息组。（长，默认：
-1) - id
-
要编制索引的文档的 ID。如果设置，则INDEX_ID标头值将基于每条消息覆盖此属性。（表达式，默认值：
<none>) - 指数
-
索引的名称。如果设置，则INDEX_NAME标头值将基于每条消息覆盖此属性。（字符串，默认：
<none>) - 路由
-
指示要路由到的分片。如果未提供，Elasticsearch 将默认为文档 ID 的哈希值。（字符串，默认：
<none>) - 超时秒
-
分片可用的超时。如果未设置，则默认为 Elasticsearch 客户端设置的 1 分钟。（长，默认：
0)
spring.elastic搜索
- 连接超时
-
与 Elasticsearch 通信时使用的连接超时。（持续时间，默认：
1s) - 密码
-
使用 Elasticsearch 进行身份验证的密码。（字符串，默认：
<none>) - 路径前缀
-
添加到发送到 Elasticsearch 的每个请求的路径中的前缀。（字符串，默认：
<none>) - 套接字保持活动状态
-
是否在客户端和 Elasticsearch 之间启用套接字保持活动状态。（布尔值，默认：
false) - 套接字超时
-
与 Elasticsearch 通信时使用的套接字超时。（持续时间，默认：
30s) - 尿素
-
要使用的 Elasticsearch 实例的逗号分隔列表。（List<String>，默认值：
[http://localhost:9200]) - 用户名
-
使用 Elasticsearch 进行身份验证的用户名。（字符串，默认：
<none>)
spring.elasticsearch.restclient.sniffer
- 故障后延迟
-
失败后计划的嗅探执行的延迟。（持续时间，默认：
1m) - 间隔
-
连续普通嗅探执行之间的间隔。（持续时间，默认：
5m)
spring.elasticsearch.restclient.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<none>)
7.3.2. 运行此接收器的示例
-
从文件夹
elasticsearch-sink:./mvnw clean package -
CD 应用程序
-
cd 到正确的 Binder 生成的应用程序（Kafka 或 RabbitMQ）
-
./mvnw clean package -
确保您正在运行 Elasticsearch。例如，您可以使用以下命令将其作为 docker 容器运行。
docker run -d --name es762 -p 9200:9200 -e "discovery.type=single-node" elasticsearch:7.6.2 -
如果中间件尚未运行，请启动中间件（Kafka 或 RabbitMQ）。
-
java -jar target/elasticsearch-sink-<kafka|rabbit>-3.0.0-SNAPSHOT.jar --spring.cloud.stream.bindings.input.destination=els-in --elasticsearch.consumer.index=testing -
将一些 JSON 数据发送到中间件目标。例如：
{"foo":"bar"} -
验证数据是否已编制索引：
curl localhost:9200/testing/_search
7.4. 文件接收器
文件接收器应用将其收到的每条消息写入文件。
7.4.1. 有效载荷
-
java.io.File -
java.io.InputStream -
byte[] -
String
7.4.2. 选项
这file-sink具有以下选项：
- 文件.消费者.二进制文件
-
一个标志，指示是否应禁止在写入后添加换行符。（布尔值，默认：
false) - file.consumer.charset
-
写入文本内容时要使用的字符集。（字符串，默认：
UTF-8) - 文件.消费者.目录
-
目标文件的父目录。（文件，默认：
<none>) - 文件.消费者.目录表达式
-
要计算目标文件的父目录的表达式。（字符串，默认：
<none>) - 文件.消费者.mode
-
如果目标文件已存在，则要使用的 FileExistsMode。（FileExistsMode，默认：
<none>，可能的值：APPEND,APPEND_NO_FLUSH,FAIL,IGNORE,REPLACE,REPLACE_IF_MODIFIED) - file.consumer.name
-
目标文件的名称。（字符串，默认：
file-consumer) - 文件.消费者.名称表达式
-
要计算目标文件名称的表达式。（字符串，默认：
<none>) - file.consumer.suffix
-
要附加到文件名的后缀。（字符串，默认：
<empty string>)
7.5. FTP 接收器
FTP 接收器是将文件从传入邮件推送到 FTP 服务器的简单选项。
它使用ftp-outbound-adapter，因此传入消息可以是java.io.File对象，一个String（文件内容）
或数组bytes（文件内容也是如此）。
要使用此接收器，您需要一个用户名和一个密码才能登录。
默认情况下，Spring Integration 将使用o.s.i.file.DefaultFileNameGenerator如果未指定任何内容。DefaultFileNameGenerator将确定文件名
基于file_name标头（如果存在）在MessageHeaders，或者如果Message已经是一个java.io.File，那么它会
使用该文件的原始名称。 |
7.5.1. 标头
-
file_name（见上文注释）
7.5.2. 有效载荷
-
java.io.File -
java.io.InputStream -
byte[] -
String
7.5.3. 输出
N/A（写入 FTP 服务器）。
7.5.4. 选项
ftp 接收器具有以下选项：
按前缀分组的属性：
ftp.consumer 的
- 自动创建目录
-
是否创建远程目录。（布尔值，默认：
true) - 文件名表达式
-
用于生成远程文件名的 SpEL 表达式。（字符串，默认：
<none>) - 模式
-
如果远程文件已存在，则要执行的作。（FileExistsMode，默认：
<none>，可能的值：APPEND,APPEND_NO_FLUSH,FAIL,IGNORE,REPLACE,REPLACE_IF_MODIFIED) - 远程目录
-
远程 FTP 目录。（字符串，默认：
/) - 远程文件分隔符
-
远程文件分隔符。（字符串，默认：
/) - 临时远程目录
-
如果 '#isUseTemporaryFilename（）' 为 true，则将写入文件的临时目录。（字符串，默认：
/) - tmp-文件后缀
-
传输过程中要使用的后缀。（字符串，默认：
.tmp) - 使用临时文件名
-
是否写入临时文件并重命名。（布尔值，默认：
true)
ftp.工厂
- 缓存会话
-
缓存会话。（布尔值，默认：
<none>) - 客户端模式
-
用于 FTP 会话的客户端模式。（ClientMode，默认：
<none>，可能的值：ACTIVE,PASSIVE) - 主机
-
服务器的主机名。（字符串，默认：
localhost) - 密码
-
用于连接到服务器的密码。（字符串，默认：
<none>) - 端口
-
服务器的端口。（整数，默认：
21) - 用户名
-
用于连接到服务器的用户名。（字符串，默认：
<none>)
7.6. JDBC 接收器
JDBC 接收器允许您将传入的有效负载持久化到 RDBMS 数据库中。
这jdbc.consumer.columns属性表示COLUMN_NAME[:EXPRESSION_FOR_VALUE]哪里EXPRESSION_FOR_VALUE（连同冒号）是可选的。
在这种情况下，值是通过生成的表达式计算的，例如payload.COLUMN_NAME，因此这样我们就可以从对象属性直接映射到表列。
例如，我们有一个 JSON 有效负载，如下所示：
{
"name": "My Name",
"address": {
"city": "Big City",
"street": "Narrow Alley"
}
}因此，我们可以将其插入到表格中name,city和street结构使用配置：
--jdbc.consumer.columns=name,city:address.city,street:address.street此接收器支持批量插入，只要基础 JDBC 驱动程序支持。
批次插入通过batch-size和idle-timeout性能：
传入消息将聚合到batch-size消息存在，然后作为批处理插入。
如果idle-timeout毫秒过去没有新消息，则插入聚合的批次，即使它小于batch-size，限制最大延迟。
该模块还使用 Spring Boot 的 DataSource 支持来配置数据库连接，因此像spring.datasource.url 等适用。 |
7.6.1. 示例
java -jar jdbc-sink.jar --jdbc.consumer.tableName=names \
--jdbc.consumer.columns=name \
--spring.datasource.driver-class-name=org.mariadb.jdbc.Driver \
--spring.datasource.url='jdbc:mysql://localhost:3306/test7.6.2. 有效载荷
7.6.3. 选项
jdbc 接收器具有以下选项：
按前缀分组的属性：
jdbc.consumer
- 批量大小
-
将数据刷新到数据库表时消息数的阈值。（整数，默认：
1) - 列
-
逗号分隔了基于冒号的列名和 SpEL 表达式对，用于插入/更新的值。初始化时使用名称来发出 DDL。（字符串，默认：
payload:payload.toString()) - 空闲超时
-
数据自动刷新到数据库表时的空闲超时（以毫秒为单位）。（长，默认：
-1) - 初始化
-
'true'、'false' 或表的自定义初始化脚本的位置。（字符串，默认：
false) - 表名
-
要写入的表的名称。（字符串，默认：
messages)
spring.datasource
- 驱动程序类名称
-
JDBC 驱动程序的完全限定名称。默认情况下，根据 URL 自动检测。（字符串，默认：
<none>) - 密码
-
数据库的登录密码。（字符串，默认：
<none>) - 网址
-
数据库的 JDBC URL。（字符串，默认：
<none>) - 用户名
-
数据库的登录用户名。（字符串，默认：
<none>)
7.7. Apache Kafka 接收器
此模块将消息发布到 Apache Kafka。
7.7.1. 选项
kafka 接收器具有以下选项：
（有关 Spring for Apache Kafka 配置属性，请参阅 Spring Boot 文档）
按前缀分组的属性：
卡夫卡.publisher
- 钥匙
-
Kafka 记录键 - 由 keyExpression 覆盖（如果提供）。（字符串，默认：
<none>) - 键表达式
-
计算结果为 Kafka 记录键的 SpEL 表达式。（表达式，默认值：
<none>) - 映射标头
-
将要映射的标头。（String[]，默认：
[*]) - 分区
-
Kafka 主题分区 - 由 partitionExpression 覆盖（如果提供）。（整数，默认：
<none>) - 分区表达式
-
计算结果为 Kafka 主题分区的 SpEL 表达式。（表达式，默认值：
<none>) - 发送超时
-
Kafka 生产者处理程序应等待发送作结果多长时间。默认为 10 秒。（持续时间，默认：
10s) - 同步
-
如果 Kafka 生产者处理程序应在同步模式下运行，则为 True。（布尔值，默认：
false) - 时间戳
-
Kafka 记录时间戳 - 由 timestampExpression 覆盖（如果提供）。（长，默认：
<none>) - 时间戳表达式
-
计算结果为 Kafka 记录时间戳的 SpEL 表达式。（表达式，默认值：
<none>) - 主题
-
Kafka topic - 由 topicExpression 覆盖（如果提供）。默认为 KafkaTemplate.getDefaultTopic（）（字符串，默认：
<none>) - 主题表达式
-
计算结果为 Kafka 主题的 SpEL 表达式。（表达式，默认值：
<none>) - 使用模板转换器
-
是否使用模板的消息转换器创建 Kafka 记录。（布尔值，默认：
false)
弹簧.kafka
- 引导服务器
-
用于建立与 Kafka 集群的初始连接的 host：port 对的逗号分隔列表。除非被覆盖，否则适用于所有组件。（List<String>，默认值：
<none>) - 客户端 ID
-
发出请求时要传递给服务器的 ID。用于服务器端日志记录。（字符串，默认：
<none>) - 性能
-
生产者和使用者通用的其他属性，用于配置客户端。（Map<String， String>，默认：
<none>)
spring.kafka.producer
- 确认
-
生产者要求领导者在认为请求完成之前已收到的确认数。（字符串，默认：
<none>) - 批量大小
-
默认批量大小。较小的批量大小将使批处理不那么常见，并可能降低吞吐量（批量大小为零将完全禁用批处理）。（DataSize，默认：
<none>) - 引导服务器
-
用于建立与 Kafka 集群的初始连接的 host：port 对的逗号分隔列表。覆盖生产者的全局属性。（List<String>，默认值：
<none>) - 缓冲内存
-
生产者可用于缓冲等待发送到服务器的记录的总内存大小。（DataSize，默认：
<none>) - 客户端 ID
-
发出请求时要传递给服务器的 ID。用于服务器端日志记录。（字符串，默认：
<none>) - 压缩式
-
生产者生成的所有数据的压缩类型。（字符串，默认：
<none>) - 密钥序列化器
-
序列化程序类。（Class<？>，默认：
<none>) - 性能
-
用于配置客户端的其他特定于生产者的属性。（Map<String， String>，默认：
<none>) - 重试
-
当大于零时，允许重试失败的发送。（整数，默认：
<none>) - 交易 ID 前缀
-
当非空时，为生产者启用事务支持。（字符串，默认：
<none>) - 值序列化器
-
Serializer 类。（Class<？>，默认：
<none>)
spring.kafka.模板
- 默认主题
-
将消息发送到的默认主题。（字符串，默认：
<none>) - 交易 ID 前缀
-
事务 ID 前缀，覆盖生产者工厂中的事务 ID 前缀。（字符串，默认：
<none>)
7.8. 原木汇
这logSink 使用应用程序记录器输出数据以供检查。
敬请谅解log接收器使用无类型处理程序，这会影响实际日志记录的执行方式。
这意味着，如果内容类型是文本的，则原始有效负载字节将转换为 String，否则将记录原始字节。
请参阅用户指南中的更多信息。
7.8.1. 选项
日志接收器具有以下选项：
- 日志表达式
-
一个 SpEL 表达式（针对传入消息）以计算为记录的消息。（字符串，默认：
payload) - 日志级别
-
记录消息的级别。（级别，默认：
<none>，可能的值：FATAL,ERROR,WARN,INFO,DEBUG,TRACE) - log.name
-
要使用的记录器的名称。（字符串，默认：
<none>)
7.9. MongoDB 接收器
此接收器应用程序将传入数据摄取到 MongoDB 中。
此应用程序完全基于MongoDataAutoConfiguration，因此请参阅 Spring Boot MongoDB 支持 了解更多信息。
7.9.1. 输入
有效载荷
-
任何 POJO
-
String -
byte[]
7.9.2. 选项
mongodb 接收器具有以下选项：
按前缀分组的属性：
mongodb.consumer
- 收集
-
用于存储数据的 MongoDB 集合。（字符串，默认：
<none>) - 集合表达式
-
用于评估 MongoDB 集合的 SpEL 表达式。（表达式，默认值：
<none>)
spring.data.mongodb
- 附加主机
-
其他服务器主机。不能使用 URI 设置，或者如果未指定“host”。其他主机将使用默认的 mongo 端口 27017。如果要使用不同的端口，可以使用“host：port”语法。（List<String>，默认值：
<none>) - 身份验证数据库
-
身份验证数据库名称。（字符串，默认：
<none>) - 自动索引创建
-
是否启用自动索引创建。（布尔值，默认：
<none>) - 数据库
-
数据库名称。覆盖 URI 中的数据库。（字符串，默认：
<none>) - 字段命名策略
-
要使用的 FieldNamingStrategy 的完全限定名称。（Class<？>，默认：
<none>) - 主机
-
Mongo 服务器主机。不能使用 URI 设置。（字符串，默认：
<none>) - 密码
-
mongo 服务器的登录密码。不能使用 URI 设置。（Character[]，默认：
<none>) - 端口
-
Mongo 服务器端口。不能使用 URI 设置。（整数，默认：
<none>) - 副本集名称
-
集群所需的副本集名称。不能使用 URI 设置。（字符串，默认：
<none>) - URI
-
Mongo 数据库 URI。覆盖主机、端口、用户名和密码。（字符串，默认：
mongodb://localhost/test) - 用户名
-
mongo 服务器的登录用户。不能使用 URI 设置。（字符串，默认：
<none>) - uuid-表示
-
将 UUID 转换为 BSON 二进制值时使用的表示。（UuidRepresentation，默认：
java-legacy，可能的值：UNSPECIFIED,STANDARD,C_SHARP_LEGACY,JAVA_LEGACY,PYTHON_LEGACY)
spring.data.mongodb.gridfs
- 桶
-
GridFS 存储桶名称。（字符串，默认：
<none>) - 数据库
-
GridFS 数据库名称。（字符串，默认：
<none>)
spring.data.mongodb.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<none>) - 启用
-
是否启用 SSL 支持。除非另有说明，否则如果提供“捆绑包”，则自动启用。（布尔值，默认：
<none>)
7.10. MQTT 接收器
该模块向 MQTT 发送消息。
7.10.1. 有效载荷：
-
byte[] -
String
7.10.2. 选项
mqtt 接收器具有以下选项：
按前缀分组的属性：
MQTT
- 清洁会话
-
客户端和服务器是否应记住重新启动和重新连接的状态。（布尔值，默认：
true) - 连接超时
-
连接超时（以秒为单位）。（整数，默认：
30) - 保持活动间隔
-
ping间隔（以秒为单位）。（整数，默认：
60) - 密码
-
连接到代理时要使用的密码。（字符串，默认：
guest) - 坚持
-
'内存'或'文件'。（字符串，默认：
memory) - 持久性目录
-
持久性目录。（字符串，默认：
/tmp/paho) - SSL-属性
-
MQTT 客户端 SSL 属性。（Map<String， String>，默认：
<none>) - 网址
-
MQTT 代理的位置（逗号分隔的列表）。（String[]，默认：
[tcp://localhost:1883]) - 用户名
-
连接到代理时要使用的用户名。（字符串，默认：
guest)
mqtt.consumer 的
- 异步
-
是否使用异步发送。（布尔值，默认：
false) - 字符集
-
用于将 String 有效负载转换为 byte[] 的字符集。（字符串，默认：
UTF-8) - 客户端 ID
-
标识客户端。（字符串，默认：
stream.client.id.sink) - QOS
-
使用的服务质量。（整数，默认：
1) - 保留
-
是否设置“保留”标志。（布尔值，默认：
false) - 主题
-
接收器将发布到的主题。（字符串，默认：
stream.mqtt)
7.11. Pgcopy 接收器
使用 PostgreSQL COPY 命令将其传入有效负载写入 RDBMS 的模块。
7.11.1. 输入
头
有效载荷
-
任何
列表达式将根据消息进行计算，并且表达式通常仅与一种类型（例如 Map 或 bean 等）兼容。
7.11.2. 输出
不适用
7.11.3. 选项
jdbc 接收器具有以下选项：
- spring.datasource.driver-class-name
-
JDBC 驱动程序的完全限定名称。默认情况下，根据 URL 自动检测。（字符串，默认：
<none>) - spring.datasource.password
-
数据库的登录密码。（字符串，默认：
<none>) - spring.datasource.url
-
数据库的 JDBC URL。（字符串，默认：
<none>) - spring.datasource.用户名
-
数据库的登录用户名。（字符串，默认：
<none>)
该模块还使用 Spring Boot 的 DataSource 支持来配置数据库连接，因此像spring.datasource.url 等适用。 |
7.11.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps您可以在此处找到相应的基于活页夹的项目。 然后，您可以 cd 到其中一个文件夹中并构建它：
$ ./mvnw clean package要运行集成测试，请在本地主机上启动 PostgreSQL 数据库：
docker run -e POSTGRES_PASSWORD=spring -e POSTGRES_DB=test -p 5432:5432 -d postgres:latest7.11.5. 示例
java -jar pgcopy-sink.jar --tableName=names --columns=name --spring.datasource.driver-class-name=org.mariadb.jdbc.Driver \
--spring.datasource.url='jdbc:mysql://localhost:3306/test7.12. RabbitMQ Sink
该模块向 RabbitMQ 发送消息。
7.12.1. 选项
兔子水槽有以下选项：
（有关 RabbitMQ 连接属性，请参阅 Spring Boot 文档）
按前缀分组的属性：
兔
- 转换器 bean-name
-
自定义消息转换器的 bean 名称;如果省略，则使用 SimpleMessageConverter。如果是 'jsonConverter'，则将为您创建一个 Jackson2JsonMessageConverter bean。（字符串，默认：
<none>) - 交换
-
Exchange 名称 - 由 exchangeNameExpression（如果提供）覆盖。（字符串，默认：
<empty string>) - 交换表达式
-
计算结果为交换名称的 SpEL 表达式。（表达式，默认值：
<none>) - 标头映射最后
-
映射出站邮件的标头时，请确定是在转换邮件之前还是之后映射标头。（布尔值，默认：
true) - 映射请求标头
-
将要映射的标头。（String[]，默认：
[*]) - 自己的连接
-
如果为 true，请根据启动属性使用单独的连接。（布尔值，默认：
false) - 持久交付模式
-
“amqp_deliveryMode”标头不存在时的默认传递模式，对于 PERSISTENT。（布尔值，默认：
false) - 路由键
-
路由键 - 由 routingKeyExpression（如果提供）覆盖。（字符串，默认：
<none>) - 路由键表达式
-
计算结果为路由键的 SpEL 表达式。（表达式，默认值：
<none>)
spring.rabbitmq
- 地址随机模式
-
用于打乱已配置地址的模式。（AddressShuffleMode，默认：
none，可能的值：NONE,RANDOM,INORDER) - 地址
-
客户端应连接到的地址的逗号分隔列表。设置后，将忽略主机和端口。（字符串，默认：
<none>) - 通道 rpc 超时
-
通道中 RPC 调用的延续超时。将其设置为零以永久等待。（持续时间，默认：
10m) - 连接超时
-
连接超时。将其设置为零以永久等待。（持续时间，默认：
<none>) - 主机
-
RabbitMQ 主机。如果设置了地址，则忽略。（字符串，默认：
localhost) - 密码
-
登录以对代理进行身份验证。（字符串，默认：
guest) - 端口
-
RabbitMQ 端口。如果设置了地址，则忽略。默认为 5672，如果启用了 SSL，则默认为 5671。（整数，默认：
<none>) - 发布者确认类型
-
确认使用的发布者类型。（ConfirmType，默认：
<none>，可能的值：SIMPLE,CORRELATED,NONE) - 发布者返回
-
是否启用发布商退货。（布尔值，默认：
false) - 请求通道最大值
-
客户端请求的每个连接的通道数。使用 0 表示无限制。（整数，默认：
2047) - 请求心跳
-
请求的检测信号超时;零表示无。如果未指定持续时间后缀，则将使用秒。（持续时间，默认：
<none>) - 用户名
-
登录用户以向代理进行身份验证。（字符串，默认：
guest) - 虚拟主机
-
连接到代理时要使用的虚拟主机。（字符串，默认：
<none>)
spring.rabbitmq.模板
- 默认接收队列
-
当未显式指定消息时，要从中接收消息的默认队列的名称。（字符串，默认：
<none>) - 交换
-
用于发送作的默认交换的名称。（字符串，默认：
<empty string>) - 命令的
-
是否启用强制消息。（布尔值，默认：
<none>) - 接收超时
-
receive（）作超时。（持续时间，默认：
<none>) - 回复超时
-
sendAndReceive（）作超时。（持续时间，默认：
<none>) - 路由键
-
用于发送作的默认路由键的值。（字符串，默认：
<empty string>)
7.13. Redis 接收器
向 Redis 发送消息。
7.13.1. 选项
Redis 接收器具有以下选项：
按前缀分组的属性：
redis.consumer
- 钥匙
-
存储到键时要使用的文字键名称。（字符串，默认：
<none>) - 键表达式
-
用于存储到键的 SpEL 表达式。（字符串，默认：
<none>) - 队列
-
存储在队列中时要使用的文字队列名称。（字符串，默认：
<none>) - 队列表达式
-
用于队列的 SpEL 表达式。（字符串，默认：
<none>) - 主题
-
发布到主题时要使用的文字主题名称。（字符串，默认：
<none>) - 主题表达式
-
用于主题的 SpEL 表达式。（字符串，默认：
<none>)
spring.data.redis
- 客户端名称
-
要在与 CLIENT SETNAME 的连接上设置的客户端名称。（字符串，默认：
<none>) - 客户端类型
-
要使用的客户端类型。默认情况下，根据类路径自动检测。（ClientType，默认：
<none>，可能的值：LETTUCE,JEDIS) - 连接超时
-
连接超时。（持续时间，默认：
<none>) - 数据库
-
连接工厂使用的数据库索引。（整数，默认：
0) - 主机
-
Redis 服务器主机。（字符串，默认：
localhost) - 密码
-
redis 服务器的登录密码。（字符串，默认：
<none>) - 端口
-
Redis 服务器端口。（整数，默认：
6379) - 超时
-
读取超时。（持续时间，默认：
<none>) - 网址
-
连接 URL。覆盖主机、端口、用户名和密码。示例：redis://user：[email protected]：6379（字符串，默认：
<none>) - 用户名
-
redis 服务器的登录用户名。（字符串，默认：
<none>)
spring.data.redis.cluster
- 最大重定向
-
跨集群执行命令时要遵循的最大重定向数。（整数，默认：
<none>) - 节点
-
要从中引导的“host：port”对的逗号分隔列表。这表示集群节点的“初始”列表，并且需要至少有一个条目。（List<String>，默认值：
<none>)
spring.data.redis.jedis.pool
- 启用
-
是否启用池。如果“commons-pool2”可用，则自动启用。使用 Jedis，在哨兵模式下隐式启用池化，并且此设置仅适用于单节点设置。（布尔值，默认：
<none>) - 最大活动
-
池在给定时间可以分配的最大连接数。使用负值表示无限制。（整数，默认：
8) - 最大空闲
-
池中“空闲”连接的最大数量。使用负值表示无限数量的空闲连接。（整数，默认：
8) - 最大等待
-
在池耗尽时引发异常之前，连接分配应阻止的最长时间。使用负值无限期阻止。（持续时间，默认：
-1ms) - 最小空闲
-
池中要维护的最小空闲连接数的目标。仅当此设置和逐出运行之间的时间均为正时，此设置才有效。（整数，默认：
0) - 驱逐运行之间的时间
-
空闲对象 evictor 线程运行之间的时间。当为正时，空闲对象 evictor 线程将启动，否则不会执行空闲对象逐出。（持续时间，默认：
<none>)
spring.data.redis.lettuce.pool
- 启用
-
是否启用池。如果“commons-pool2”可用，则自动启用。使用 Jedis，在哨兵模式下隐式启用池化，并且此设置仅适用于单节点设置。（布尔值，默认：
<none>) - 最大活动
-
池在给定时间可以分配的最大连接数。使用负值表示无限制。（整数，默认：
8) - 最大空闲
-
池中“空闲”连接的最大数量。使用负值表示无限数量的空闲连接。（整数，默认：
8) - 最大等待
-
在池耗尽时引发异常之前，连接分配应阻止的最长时间。使用负值无限期阻止。（持续时间，默认：
-1ms) - 最小空闲
-
池中要维护的最小空闲连接数的目标。仅当此设置和逐出运行之间的时间均为正时，此设置才有效。（整数，默认：
0) - 驱逐运行之间的时间
-
空闲对象 evictor 线程运行之间的时间。当为正时，空闲对象 evictor 线程将启动，否则不会执行空闲对象逐出。（持续时间，默认：
<none>)
spring.data.redis.lettuce
- 关机超时
-
关机超时。（持续时间，默认：
100ms)
spring.data.redis.sentinel
- 主人
-
Redis 服务器的名称。（字符串，默认：
<none>) - 节点
-
逗号分隔的“host：port”对列表。（List<String>，默认值：
<none>) - 密码
-
用于向哨兵进行身份验证的密码。（字符串，默认：
<none>) - 用户名
-
用于向哨兵进行身份验证的登录用户名。（字符串，默认：
<none>)
spring.data.redis.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<none>) - 启用
-
是否启用 SSL 支持。除非另有说明，否则如果提供“捆绑包”，则自动启用。（布尔值，默认：
<none>)
7.14. 路由器接收器
此应用程序将消息路由到命名通道。
7.14.1. 选项
路由器接收器具有以下选项：
- router.default-输出绑定
-
在何处发送不可路由的消息。（字符串，默认：
<none>) - router.destination-mappings
-
目标映射作为以名称-值对为分隔的新行字符串，例如 'foo=bar\n baz=car'。（属性，默认值：
<none>) - router.expression
-
要应用于消息以确定要路由到的通道的表达式。请注意，text、json 或 xml 等内容类型的有效负载线格式是 byte[] 而不是 String！。有关如何处理字节数组有效负载内容，请参阅文档。（表达式，默认值：
<none>) - router.refresh-delay
-
检查脚本更改的频率（以毫秒为单位）（如果存在）;< 0 表示不刷新。（整数，默认：
60000) - 需要 router.resolution-required
-
是否需要通道分辨率。（布尔值，默认：
false) - 路由器.脚本
-
返回通道或通道映射分辨率键的 groovy 脚本的位置。（资源，默认值：
<none>) - router.变量
-
变量绑定为换行符，以名称-值对为分隔字符串，例如 'foo=bar\n baz=car'。（属性，默认值：
<none>) - router.变量位置
-
包含自定义脚本变量绑定的属性文件的位置。（资源，默认值：
<none>)
此路由器接收器基于StreamBridgeAPI 来自 Spring Cloud Stream，因此可以根据需要创建目标。
在这种情况下，一个defaultOutputBinding只有当 key 未包含在destinationMappings.
这resolutionRequired = true忽略defaultOutputBinding如果没有映射并且尚未声明相应的绑定，则会抛出异常。 |
您可以使用spring.cloud.stream.dynamicDestinations财产。
默认情况下，所有已解析的目的地都将动态绑定;如果此属性具有逗号分隔的目标名称列表，则仅绑定这些名称。
解析到不在此列表中的目标的消息将被路由到defaultOutputBinding，也必须出现在列表中。
这destinationMappings用于将评估结果映射到实际的目标名称。
7.14.2. 基于 SpEL 的路由
该表达式根据消息进行计算，并返回通道名称或通道名称映射的键。
有关更多信息，请参阅 Spring Integration 参考手册配置通用路由器部分中的“路由器和 Spring 表达式语言 （SpEL）”小节。
7.14.3. 基于 Groovy 的路由
除了 SpEL 表达式，还可以使用 Groovy 脚本。让我们在文件系统中的“file：/my/path/router.groovy”或“classpath：/my/path/router.groovy”创建一个 Groovy 脚本：
println("Groovy processing payload '" + payload + "'")
if (payload.contains('a')) {
return "foo"
}
else {
return "bar"
}
如果要将变量值传递给脚本，可以使用 variables 选项静态绑定值，或者选择使用 propertiesLocation 选项将路径传递到包含绑定的属性文件。 文件中的所有属性都将作为变量提供给脚本。您可以同时指定变量和 propertiesLocation，在这种情况下，作为变量提供的任何重复值都会覆盖 propertiesLocation 中提供的值。 请注意，有效负载和标头是隐式绑定的，以允许您访问消息中包含的数据。
有关更多信息，请参阅 Spring Integration 参考手册 Groovy 支持。
7.15. RSocket 水槽
RSocket 接收器，使用 RSocket 协议的触发并忘记策略发送数据。
7.15.1. 选项
rsocket 接收器具有以下选项：
- rsocket.consumer.host
-
RSocket 主机。（字符串，默认：
localhost) - rsocket.consumer.port
-
RSocket 端口。（整数，默认：
7000) - rsocket.consumer.route
-
用于 RSocket 的路由。（字符串，默认：
<none>) - rsocket.consumer.uri 的
-
可用于基于 websocket 的传输的 URI。（URI，默认：
<none>)
7.16. 亚马逊 S3 接收器
此接收器应用程序支持将对象传输到 Amazon S3 存储桶。
文件有效负载（和目录递归）将传输到remote目录（S3 存储桶）到local部署应用程序的目录。
此接收器接受的消息必须包含payload如：
-
File，包括递归上传的目录; -
InputStream; -
byte[]
7.16.1. 选项
s3 接收器具有以下选项：
按前缀分组的属性：
s3.消费者
- 前交叉韧带
-
S3 对象访问控制列表。（ObjectCannedACL，默认：
<none>，可能的值：private,public-read,public-read-write,authenticated-read,aws-exec-read,bucket-owner-read,bucket-owner-full-control,null) - ACL 表达
-
用于评估 S3 对象访问控制列表的表达式。（表达式，默认值：
<none>) - 桶
-
用于存储目标文件的 AWS 存储桶。（字符串，默认：
<none>) - 桶表达式
-
用于评估 AWS 存储桶名称的表达式。（表达式，默认值：
<none>) - 键表达式
-
用于计算 S3 对象键的表达式。（表达式，默认值：
<none>)
spring.cloud.aws.凭证
- 访问密钥
-
要与静态提供程序一起使用的访问密钥。（字符串，默认：
<none>) - 实例配置文件
-
配置实例配置文件凭据提供程序，无需进一步配置。（布尔值，默认：
false) - 轮廓
-
AWS 配置文件。（配置文件，默认：
<none>) - 密钥
-
要与静态提供程序一起使用的密钥。（字符串，默认：
<none>)
spring.cloud.aws.region
- 实例配置文件
-
配置实例配置文件区域提供程序，无需进一步配置。（布尔值，默认：
false) - 轮廓
-
AWS 配置文件。（配置文件，默认：
<none>) - 静态的
-
<缺少文档>（字符串，默认：
<none>)
spring.cloud.aws.s3
- 启用加速模式
-
在访问 S3 时启用使用加速端点的选项。Accelerate 终端节点允许使用 Amazon CloudFront 的全球分布式边缘站点更快地传输对象。（布尔值，默认：
<none>) - 启用校验和验证
-
禁用对存储在 S3 中的对象的校验和进行验证的选项。（布尔值，默认：
<none>) - 已启用分块编码
-
在签署 {@link software.amazon.awssdk.services.s3.model.PutObjectRequest} 和 {@link software.amazon.awssdk.services.s3.model.UploadPartRequest} 的请求负载时启用使用分块编码的选项。（布尔值，默认：
<none>) - 跨区域启用
-
启用跨区域存储桶访问。（布尔值，默认：
<none>) - 端点
-
覆盖默认终结点。（URI，默认：
<none>) - 路径样式访问启用
-
启用使用路径样式访问而不是 DNS 样式访问的选项。DNS 样式的访问是首选，因为它在访问 S3 时可以实现更好的负载平衡。（布尔值，默认：
<none>) - 地区
-
覆盖默认区域。（字符串，默认：
<none>) - 使用arn-region-enabled
-
如果 S3 资源 ARN 作为 S3作的目标传入，该作的区域与配置客户端的区域不同，则必须将此标志设置为“true”，以允许客户端对 ARN 中指定的区域进行跨区域调用，否则将引发异常。（布尔值，默认：
<none>)
spring.cloud.aws.s3.crt
- 初始读取缓冲区大小（以字节为单位）
-
配置客户端将用于缓冲从 S3 下载的部件的起始缓冲区大小。保持更大的窗口以保持高下载吞吐量;除非窗口足够大以容纳多个部分，否则无法并行下载部件。保持较小的窗口以限制内存中缓冲的数据量。（长，默认：
<none>) - 最大并发
-
指定传输期间应建立的最大 S3 连接数。（整数，默认：
<none>) - 最小部分大小（字节）
-
设置传输部件的最小部件大小。减小最小部件大小会导致多部件传输被拆分为更多较小的部件。将此值设置得太低会对传输速度产生负面影响，从而导致每个部件的额外延迟和网络通信。（长，默认：
<none>) - 目标吞吐量（以 Gbps 为单位）
-
传输请求的目标吞吐量。值越高，表示将打开更多 S3 连接。传输管理器是否能够实现配置的目标吞吐量取决于环境的网络带宽和配置的“最大并发”等多种因素。（双精度，默认：
<none>)
目标生成的应用程序基于AmazonS3SinkConfiguration可以通过S3MessageHandler.UploadMetadataProvider和/或S3ProgressListener，这些被注入S3MessageHandler豆。 有关更多详细信息，请参阅 Spring Integration AWS 支持。
7.16.2. Amazon AWS 通用选项
Amazon S3 Sink（与所有其他 Amazon AWS 应用程序一样）基于 Spring Cloud AWS 项目作为基础，其自动配置类由 Spring Boot 自动使用。有关所需和有用的自动配置属性，请参阅他们的文档。
其中一些是关于 AWS 凭证的：
-
spring.cloud.aws.credentials.accessKey
-
spring.cloud.aws.credentials.secretKey
-
spring.cloud.aws.credentials.instanceProfile
-
spring.cloud.aws.credentials.profileName
-
spring.cloud.aws.credentials.profile路径
其他适用于 AWSRegion定义：
-
cloud.aws.region.auto
-
cloud.aws.region.static
例子
java -jar s3-sink.jar --s3.bucket=/tmp/bar7.17. SFTP 接收器
SFTP 接收器是一个简单的选项，用于将文件从传入邮件推送到 SFTP 服务器。
它使用sftp-outbound-adapter，因此传入消息可以是java.io.File对象，一个String（文件内容）
或数组bytes（文件内容也是如此）。
要使用此接收器，您需要一个用户名和一个密码才能登录。
默认情况下，Spring Integration 将使用o.s.i.file.DefaultFileNameGenerator如果未指定任何内容。DefaultFileNameGenerator将确定文件名
基于file_name标头（如果存在）在MessageHeaders，或者如果Message已经是一个java.io.File，那么它会
使用该文件的原始名称。 |
配置sftp.factory.known-hosts-expression选项，则评估的根对象是应用程序上下文，例如可能是sftp.factory.known-hosts-expression = @systemProperties['user.home'] + '/.ssh/known_hosts'.
7.17.1. 输入
头
-
file_name（见上文注释）
有效载荷
-
java.io.File -
java.io.InputStream -
byte[] -
String
7.17.2. 输出
N/A（写入 SFTP 服务器）。
7.17.3. 选项
sftp 接收器具有以下选项：
按前缀分组的属性：
sftp.消费者
- 自动创建目录
-
是否创建远程目录。（布尔值，默认：
true) - 文件名表达式
-
用于生成远程文件名的 SpEL 表达式。（字符串，默认：
<none>) - 模式
-
如果远程文件已存在，则要执行的作。（FileExistsMode，默认：
<none>，可能的值：APPEND,APPEND_NO_FLUSH,FAIL,IGNORE,REPLACE,REPLACE_IF_MODIFIED) - 远程目录
-
远程 FTP 目录。（字符串，默认：
/) - 远程文件分隔符
-
远程文件分隔符。（字符串，默认：
/) - 临时远程目录
-
如果 'isUseTemporaryFilename（）' 为 true，则将写入文件的临时目录。（字符串，默认：
/) - tmp-文件后缀
-
传输过程中要使用的后缀。（字符串，默认：
.tmp) - 使用临时文件名
-
是否写入临时文件并重命名。（布尔值，默认：
true)
sftp.消费者.工厂
- 允许未知键
-
True 表示允许未知或更改的键。（布尔值，默认：
false) - 缓存会话
-
缓存会话。（布尔值，默认：
<none>) - 主机
-
服务器的主机名。（字符串，默认：
localhost) - 已知主机表达式
-
解析为已知主机文件位置的 SpEL 表达式。（表达式，默认值：
<none>) - 密码短语
-
用户私钥的密码。（字符串，默认：
<empty string>) - 密码
-
用于连接到服务器的密码。（字符串，默认：
<none>) - 端口
-
服务器的端口。（整数，默认：
22) - 私钥
-
用户私钥的资源位置。（资源，默认值：
<none>) - 用户名
-
用于连接到服务器的用户名。（字符串，默认：
<none>)
7.18. TCP接收器
此模块使用编码器将消息写入 TCP。
TCP 是一种流式处理协议，需要某种机制来构建在线路上的消息。许多编码器是 available，默认值为“CRLF”。
7.18.1. 选项
tcp 接收器具有以下选项：
按前缀分组的属性：
tcp.consumer
- 字符集
-
从字节转换为字符串时使用的字符集。（字符串，默认：
UTF-8) - 关闭
-
是否在每条消息后关闭套接字。（布尔值，默认：
false) - 编码器
-
发送消息时要使用的编码器。（编码，默认：
<none>，可能的值：CRLF,LF,NULL,STXETX,RAW,L1,L2,L4) - 主机
-
此接收器将连接到的主机。（字符串，默认：
<none>)
TCP
- 蔚来
-
是否使用蔚来。（布尔值，默认：
false) - 端口
-
要收听的端口;0 表示作系统选择端口。（整数，默认：
1234) - 反向查找
-
对远程 IP 地址执行反向 DNS 查找;如果为 false，则邮件标头中仅包含 IP 地址。（布尔值，默认：
false) - 套接字超时
-
未收到数据时关闭套接字之前的超时 （ms）。（整数，默认：
120000) - 使用直接缓冲区
-
是否使用直接缓冲区。（布尔值，默认：
false)
7.18.2. 可用的编码器
- CRLF（默认）
-
以回车符 （0x0d） 结尾的文本，后跟换行符 （0x0a）
- 如果
-
以换行符结尾的文本 （0x0a）
- 零
-
以空字节 （0x00） 结尾的文本
- STXETX
-
前面是 STX （0x02） 并以 ETX （0x03） 结尾的文本
- 生
-
no structure - 客户端通过关闭套接字来指示完整的消息
- L1
-
前面有一个 1 字节（无符号）长度字段的数据（最多支持 255 字节）
- L2
-
数据前面有一个两个字节（无符号）长度字段（最多 216-1 字节）
- L4
-
数据前面有一个四字节（有符号）长度字段（最多 231-1 字节）
7.19. 吞吐量接收器
接收器，它将按选定的时间间隔对消息进行计数并记录观察到的吞吐量。
7.19.1. 选项
吞吐量接收器具有以下选项：
- 吞吐量.report-every-ms
-
报告的频率。（整数，默认：
1000)
7.20. Twitter 消息接收器
从身份验证用户向指定用户发送直接消息。
需要 JSON POST 正文和Content-Typeheader 设置为application/json.
| 当收到来自用户的消息时，您可以在 24 小时内发送最多 5 条消息作为响应。 收到的每条消息都会重置 24 小时窗口和 5 条分配的消息。 在 24 小时窗口内发送第 6 条消息或在 24 小时窗口之外发送消息将计入速率限制。 此行为仅适用于使用 POST direct_messages/events/new 端点时。 |
SpEL 表达式用于从输入消息中计算请求参数。
7.20.1. 选项
使用单引号 （） 将'SpEL表达式属性。
例如，要设置固定消息文本，请使用text='Fixed Text'.
对于固定目标 userId，请使用userId='666'. |
- 推特.message.update.media-id
-
要与消息关联的媒体 ID。私信只能引用单个媒体 ID。（表达式，默认值：
<none>) - 推特.message.update.屏幕名称
-
向其发送直接消息的用户的屏幕名称。（表达式，默认值：
<none>) - 推特.message.update.text
-
直接消息文本。根据需要对 URL 进行编码。最大长度为 10,000 个字符。（表达式，默认值：
payload) - 推特.message.update.user-id
-
向其发送直接消息的用户的用户 ID。（表达式，默认值：
<none>)
7.21. Twitter 更新接收器
更新身份验证用户的当前文本（例如推文）。
| 对于每次更新尝试，都会将更新文本与身份验证用户最近的推文进行比较。 任何可能导致重复的尝试都将被阻止，从而导致 403 错误。 用户不能连续提交两次相同的文本。 |
虽然不受 API 的速率限制，但用户一次可以创建的推文数量受到限制。 标准 API 的更新限制为 3 小时窗口内 300 次。 如果用户发布的更新数达到当前允许的限制，此方法将返回 HTTP 403 错误。
您可以在此处找到更新 API 的详细信息：developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-update
7.21.1. 选项
按前缀分组的属性：
推特.update
- 附件网址
-
（SpEL表达式）为了使 URL 不计入扩展推文的文本正文中，请提供 URL 作为推文附件。此 URL 必须是推文永久链接或私信深层链接。任意非 Twitter URL 必须保留在文本文本中。传递给 attachment_url 参数的 URL 与推文永久链接或私信深度链接不匹配，则在创建推文时失败并导致异常。（表达式，默认值：
<none>) - 显示坐标
-
（SpEL表达式）是否在发送推文的确切坐标上放置图钉。（表达式，默认值：
<none>) - 回复状态 ID
-
（SpEL表达式）更新要回复的现有文本的 ID。注意：除非文本中提及此参数引用的推文的作者，否则此参数将被忽略。因此，您必须在更新中包含@username，其中用户名是引用推文的作者。设置 inReplyToStatusId 后，也会自动设置auto_populate_reply_metadata。稍后确保从原始推文中查找主要@mentions，并从那里添加到新推文中。随着回复链的增长，这会将@mentions附加到扩展推文的元数据中，直到达到@mentions限制。如果原始推文已被删除，回复将失败。（表达式，默认值：
<none>) - 媒体 ID
-
（SpEL表达式）要与推文关联的media_ids的逗号分隔列表。您最多可以在推文中添加 4 张照片或 1 个动画 GIF 或 1 个视频。有关上传媒体的更多详细信息，请参阅上传媒体。（表达式，默认值：
<none>) - 地点 ID
-
（SpEL表达式）世界上的一个地方。（表达式，默认值：
<none>) - 文本
-
（SpEL表达式）文本更新的文本。根据需要对 URL 进行编码。t.co 链接换行将影响字符数。默认为消息的有效负载（表达式，默认值：
payload)
推特.update.location
- 纬度
-
此推文所指位置的纬度。除非该参数在 -90.0 到 +90.0（北方为正）范围内，否则将忽略此参数。如果没有相应的 long 参数，它也会被忽略。（表达式，默认值：
<none>) - 离子
-
此推文所指位置的经度。经度的有效范围为 -180.0 到 +180.0（东为正），包括在内。如果超出该范围、如果它不是数字、如果禁用geo_enabled或没有相应的 lat 参数，则将忽略此参数。（表达式，默认值：
<none>)
7.22. 波前汇
Wavefront 接收器使用 Messages<？>，将其转换为 Wavefront 数据格式的指标，并将指标直接发送到 Wavefront 或 Wavefront 代理。
支持常见的 ETL 用例，其中必须清理、转换现有（历史）指标数据并将其存储在 Wavefront 中以进行进一步分析。
7.22.1. 选项
波前接收器具有以下选项：
- wavefront.api Tokens
-
Wavefront API 访问Tokens。（字符串，默认：
<none>) - wavefront.metric-expression
-
计算结果为度量值的 SpEL 表达式。（表达式，默认值：
<none>) - wavefront.metric-name
-
指标的名称。默认为应用程序名称。（字符串，默认：
<none>) - 波前.proxy-uri
-
Wavefront 代理的 URL。（字符串，默认：
<none>) - 波前.source
-
发出指标的唯一应用程序、主机、容器或实例。（字符串，默认：
<none>) - wavefront.tag表达式
-
与指标关联的自定义元数据的集合。点标记不能为空。键的有效字符：字母数字、连字符 （'-'）、下划线('_')、点 （'.'）。对于值，允许使用任何字符，包括空格。要包含双引号，请用反斜杠转义它，反斜杠不能是标签值中的最后一个字符。点标记键和值组合的最大允许长度为 254 个字符（255 个字符，包括分隔键和值的 '='）。如果该值较长，则拒绝并记录该点（Map<String， Expression>，默认值：
<none>) - wavefront.timestamp 表达式
-
计算结果为指标时间戳的 SpEL 表达式（可选）。（表达式，默认值：
<none>) - 波前线.uri
-
Wavefront 环境的 URL。（字符串，默认：
<none>)
7.23. Websocket 接收器
一个简单的 Websocket Sink 实现。
7.23.1. 选项
支持以下选项：
- websocket.consumer.log级
-
netty 通道的 logLevel 。默认值为 <tt>WARN</tt>（字符串，默认：
<none>) - websocket.consumer.path
-
WebsocketSink 使用者需要连接的路径。默认值为 <tt>/websocket</tt>（字符串，默认：
/websocket) - websocket.consumer.port
-
Netty 服务器侦听的端口。默认值为 <tt>9292</tt>（整数，默认：
9292) - websocket.consumer.ssl 的
-
是否创建 {@link io.netty.handler.ssl.SslContext}。（布尔值，默认：
false) - websocket.consumer.threads
-
Netty {@link io.netty.channel.EventLoopGroup} 的线程数。默认值为 <tt>1</tt>（整数，默认：
1)
7.23.2. 示例
要验证 websocket-sink 是否接收来自其他 spring-cloud-stream 应用程序的消息，您可以使用 遵循简单的端到端设置。
第 1 步：启动 Rabbitmq
第 2 步：部署time-source
步骤 3：部署websocket-sink
最后在trace模式，以便您看到由time-source在日志中：
java -jar <spring boot application for websocket-sink> --spring.cloud.stream.bindings.input=ticktock --server.port=9393 \
--logging.level.org.springframework.cloud.fn.consumer.websocket=TRACE您应该开始在启动 WebsocketSink 的控制台中看到日志消息，如下所示：
Handling message: GenericMessage [payload=2015-10-21 12:52:53, headers={id=09ae31e0-a04e-b811-d211-b4d4e75b6f29, timestamp=1445424778065}]
Handling message: GenericMessage [payload=2015-10-21 12:52:54, headers={id=75eaaf30-e5c6-494f-b007-9d5b5b920001, timestamp=1445424778065}]
Handling message: GenericMessage [payload=2015-10-21 12:52:55, headers={id=18b887db-81fc-c634-7a9a-16b1c72de291, timestamp=1445424778066}]7.23.3. 执行器
有一个Endpoint您可以使用它来访问最后一个n发送和接收的消息。你必须
通过提供--endpoints.websocketconsumertrace.enabled=true.默认情况下，它通过host:port/websocketconsumertrace.下面是一个示例输出：
[
{
"timestamp": 1445453703508,
"info": {
"type": "text",
"direction": "out",
"id": "2ff9be50-c9b2-724b-5404-1a6305c033e4",
"payload": "2015-10-21 20:54:33"
}
},
...
{
"timestamp": 1445453703506,
"info": {
"type": "text",
"direction": "out",
"id": "2b9dbcaf-c808-084d-a51b-50f617ae6a75",
"payload": "2015-10-21 20:54:32"
}
}
]还有一个简单的 HTML 页面，您可以在其中看到文本区域中转发的消息。您可以访问
它直接通过host:port在您的浏览器中。
7.24. XMPP 接收器
“xmpp”接收器允许将消息发送到 XMPP 服务器。
7.24.1. 输入
-
byte[]
7.24.2. 输出
有效载荷
不适用
7.24.3. 选项
zeromq 接收器具有以下选项：
按前缀分组的属性：
xmpp.消费者
- 聊天
-
要向其发送消息的 XMPP 句柄。（字符串，默认：
<none>)
xmpp.factory
- 主机
-
要连接到的 XMPP 主机服务器。（字符串，默认：
<none>) - 密码
-
连接用户的密码。（字符串，默认：
<none>) - 端口
-
用于连接到主机的端口。- 默认客户端端口：5222（整数，默认：
5222) - 资源
-
要绑定到 XMPP 主机上的资源。- 可以为空，如果未设置，服务器将生成一个（字符串，默认：
<none>) - 安全模式
-
<缺少文档>（SecurityMode，默认：
<none>，可能的值：required,ifpossible,disabled) - 服务名称
-
要为 XMPP 域设置的服务名称。（字符串，默认：
<none>) - 订阅模式
-
<缺少文档>（SubscriptionMode，默认：
<none>，可能的值：accept_all,reject_all,manual) - 用户
-
连接应作为的用户。（字符串，默认：
<none>)
7.24.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps您可以在此处找到相应的基于活页夹的项目。 然后，您可以 cd 到其中一个文件夹中并构建它：
$ ./mvnw clean package7.25. ZeroMQ 接收器
“zeromq”接收器允许将消息发送到 ZeroMQ 套接字。
7.25.1. 输入
-
byte[]
7.25.2. 输出
有效载荷
不适用
7.25.3. 选项
zeromq 接收器具有以下选项：
- zeromq.consumer.connect-url
-
用于连接到 ZeroMQ 套接字的连接 URL。（字符串，默认：
<none>) - zeromq.consumer.socket 类型
-
连接应建立的套接字类型。（SocketType，默认：
<none>，可能的值：PAIR,PUB,SUB,REQ,REP,DEALER,ROUTER,PULL,PUSH,XPUB,XSUB,STREAM,CLIENT,SERVER,RADIO,DISH,CHANNEL,PEER,RAW,SCATTER,GATHER) - zeromq.consumer.topic
-
一个 Topic SpEL 表达式，用于在向订阅者发送消息之前评估主题。（表达式，默认值：
<none>)
7.25.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps您可以在此处找到相应的基于活页夹的项目。 然后，您可以 cd 到其中一个文件夹中并构建它：
$ ./mvnw clean package7.25.5. 示例
java -jar zeromq-sink.jar --zeromq.consumer.connectUrl=tcp://server:port --zeromq.consumer.topic=