应用
5. 资料来源
5.1. 德贝齐姆源
基于Debezium引擎的变更数据采集（CDC）来源。 这德贝齐姆源允许捕获数据库变更事件并通过不同的消息绑定器进行流式传输，例如阿帕奇·卡夫卡,兔子MQ以及所有春云流支持经纪人。
| 该源可与任何 Spring Cloud Stream 消息绑定器一起使用。它不受 Kafka Connect 框架限制，也不依赖于其框架。虽然这种方法具有灵活性，但存在一定的限制。 |
支持所有Debezium配置属性。只需在任何Debezium属性之前用debezium.properties。前缀。 例如，将Debezium设为connector.class财产 使用Debezium.properties.connector.class而是用源属性。
5.1.1. 数据库支持
这德贝齐姆源目前支持多种数据存储的CDC：MySQL、PostgreSQL、MongoDB、Oracle、SQL Server、Db2、Vitess 和 Spanner 数据库。
5.1.2. 选项
按前缀分组的属性：
德贝齐姆
- Debezium-native-configuration
-
<缺少文档>（属性，默认：
<没有>) - 头部格式
-
{@link ChangeEvent} 头部格式。默认为 'JSON'。（DebeziumFormat，默认：
<没有>，可能的值：JSON,阿弗罗,普罗托布夫) - 偏移-提交-策略
-
定义何时将偏移提交到偏移存储的策略。（DebeziumOffsetCommitPolicy，默认：
<没有>，可能的值：总是,周期的,默认值) - 有效载荷格式
-
{@link ChangeEvent}键和有效载荷格式。默认为“JSON”。（DebeziumFormat，默认：
<没有>，可能的值：JSON,阿弗罗,普罗托布夫) - 性能
-
Debezium 配置属性的Spring通槽包装器。所有带有 'debezium.properties.*' 前缀的属性都是原生 Debezium 属性。（Map<String， String>，默认：
<没有>)
debezium.supplier
- 复制头
-
将变更事件头复制到消息头中。（布尔值，默认：
true)
事件扁平化配置
Debezium 提供了全面的消息格式，准确地详细说明了系统中发生的变化信息。不过，有时这种格式可能不适合下游用户，因为下游用户可能需要格式化，使字段名称和值以简化的方式呈现，扁平结构。
为了简化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源运行时，它会读取源信息并定期记录补偿这决定了它处理了多少这些信息。如果源被重启，它会使用最后记录的偏移量来判断应在源信息中恢复读取的哪个位置。开箱即用，提供以下偏移存储配置选项：
-
内存内
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 指向文件的路径，偏移量存储在哪里。当 偏置.存储”设置为FileOffsetBackingStore.2 尝试提交抵消的间隔。默认是1分钟。 -
卡夫卡话题
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 卡夫卡主题的名称，用于存储偏移量。当 offset.storage设置为KafkaOffsetBackingStore.2 创建偏移存储主题时使用的分区数量。 3 创建偏移存储主题时使用的复制因子。 4 尝试提交抵消的间隔。默认是1分钟。
可以实现org.apache.kafka.connect.storage.OffsetBackingStore接口输入，提供与自定义后端键值存储绑定的偏移存储。
连接器属性
下表列出了每个连接器所有可用的Debezium性质。
这些属性可以通过在debezium.properties。前缀。
5.1.3. 示例与测试
debezium集成测试使用数据库夹具，运行在本地机器上。利用预构建的 debezium docker 数据库镜像，借助 Testcontainers 进行应用。
要从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客户端连接到数据库并创建德贝齐姆具备所需凭证的用户： |
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 数据库：
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 | 连接到运行在 的 MySQL 服务器本地主持人：3306如德贝齐姆用户。 |
| 4 | 包含变更事件值模式变更事件消息。 |
| 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 | 配置德贝齐姆源使用 PostgresConnector。 |
| 2 | 配置Debezium发动机以使用存储商店。 |
| 3 | 元数据用于识别和分发进入事件。 |
| 4 | 连接到运行于 的 PostgreSQL 服务器本地主机：5432如后文用户。 |
| 5 | 在消息中包含变更事件值的模式。 |
| 6 | 使Chage事件平坦化成为可能。 |
你也可以运行DebeziumDatabasesIntegrationTest#postgres（）使用该 Postgres 配置。
| 禁用 postgres GenericContainer 测试初始化代码。 |
MongoDB
从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'在蒙戈德终端输出，搜索类似的日志条目主持人：“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进入你的/等等/主持人
使用以下属性将Debezium Source与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 | 配置德贝齐姆源使用MongoDB连接器。 |
| 2 | 配置Debezium发动机以使用存储. |
| 3 | 与运行于 的 MongoDB 的连接本地主持：27017如德贝齐姆用户。 |
| 4 | debezium.io/docs/connectors/mongodb/#tasks |
| 5 | 包含变更事件值模式来源记录事件。 |
| 6 | 使变形事件平坦化成为可能。 |
你也可以运行DebeziumDatabasesIntegrationTest#mongodb（）使用这个MongoDB配置。
SQL Server
开始一个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-GDR2Populate with sample data form debezium SqlServer tutorial：
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 | 配置德贝齐姆源使用 SqlServerConnector。 |
| 2 | 配置Debezium发动机以使用存储国营商店。 |
| 3 | 元数据用于识别和分发进入事件。 |
| 4 | 连接到运行于 的 SQL Server本地主持：1433如南非用户。 |
你也可以运行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.supplier.mode 选项进行自定义：
-
ref 提供 java.io.file 引用
-
行 会逐行拆分文件，并为每行发送一条新消息
-
目录 默认。以字节数组的形式提供文件内容
使用--file.supplier.mode=lines你也可以提供额外的选项--file.supplier.withMarkers=true.
如果设置为 true，底层 FileSplitter 会在实际数据之前和之后发出额外的文件起始和结束标记消息。
这两个额外标记消息的有效载荷为FileSplitter.FileMarker.withMarkers 选项如果未明确设置，默认为 false。
5.2.1. 选项
文件源有以下选项：
按前缀分组的属性：
file.consumer
- markers-json
-
当“fileMarkers == true”时，指定它们应生成为FileSplitter.FileMarker对象还是JSON。（布尔值，默认：
true) - 模式
-
FileReadingMode 用于文件读取源。值包括“ref”——文件对象，“lines”——每行一条消息，或“contents”——内容以字节形式表示。（文件阅读模式，默认：
<没有>，可能的值：裁判,线,内容) - 带标记
-
设置为 true，以便在数据之前或之后发送文件开始/结束文件标记信息。仅适用于FileReadingMode的“行”。（布尔值，默认：
<没有>)
file.supplier
- 延迟-空时
-
当检测到无新文件时的延迟持续时间。（持续时间，默认：
1) - 目录
-
用于轮询新文件的目录。（文件，默认：
<没有>) - 文件名模式
-
一个简单的Ant图案来匹配文件。（字符串，默认：
<没有>) - 文件名正则表达式
-
一个正则表达式模式来匹配文件。（模式，默认：
<没有>) - 防止重复
-
设置为 true，包含一个 AcceptOnceFileListFilter 以防止重复。（布尔值，默认：
true)
Metadata.store.dynamo-db
- 创建延迟
-
创建表格重试之间的延迟。（整数，默认：
1) - 创建重试
-
创建表格请求的重试编号。（整数，默认：
25) - 读取容量
-
桌面上的读取容量。（长，默认：
1) - 桌子
-
元数据的表名。（字符串，默认：
<没有>) - 生存时间
-
表格条目的时间线。（整数，默认：
<没有>) - 写容量
-
写入容量放在桌面上。（长，默认：
1)
metadata.store.jdbc
- 地区
-
该存储库保留了消息的唯一分组标识符。（字符串，默认：
默认值) - 表前缀
-
自定义表名称的前缀。（字符串，默认：
<没有>)
Metadata.store.mongo-db
- 收集
-
MongoDB元数据集合名称。（字符串，默认：
元数据存储)
metadata.store.redis
- 钥匙
-
Redis 密钥用于元数据。（字符串，默认：
<没有>)
metadata.store
- 类型
-
表示要配置的元数据存储类型（默认为“内存”）。使用持久存储必须包含相应的 Spring Integration 依赖。（StoreType，默认：
<没有>，可能的值：蒙戈德,Redis,迪纳莫德,JDBC,Zookeeper,Hazelcast,存储)
元数据.store.zookeeper
- 连接串
-
Zookeeper 连接字符串格式为 HOST：PORT。（字符串，默认：
127.0.0.1:2181) - 编码
-
用于在Zookeeper中存储数据时使用的编码。（字符集，默认：
UTF-8) - 重试区间
-
Zookeeper作的重试间隔以毫秒计。（整数，默认：
1000) - 根
-
根节点——存储条目是该节点的子节点。（字符串，默认：
/SpringIntegration-元数据存储)
5.3. FTP源
该源应用程序支持使用FTP协议传输文件。
文件是从远程目录当地应用部署的目录。
源端发出的消息默认以字节数组形式提供。然而，这可以是
通过--模式选择：
-
裁判提供
java.io.file参考 -
线会逐行拆分文件，并为每行发送一条新消息
-
内容默认。以字节数组的形式提供文件内容
使用--模式=线你也可以提供额外的选项--withMarkers=真.
如果设置为true， 基础文件分拆器会在实际数据之前和之后发送额外的文件开头和结束标记信息。
这两个额外标记消息的有效载荷为FileSplitter.FileMarker.选项withMarkers默认为false如果没有明确设置。
另见元数据存储选项，了解可能的共享持久存储配置，以防止重启时出现重复消息。
5.3.1. 输入
N/A（从FTP服务器获取文件）。
5.3.2. 输出
模式 = 目录
头：
-
内容类型：application/octet-stream -
file_originalFile： <java.io.File> -
file_name：<文件名>
有效载荷：
一个字节[]装满了文件内容。
模式 = 线
头：
-
内容类型：文本/纯文字 -
file_orginalFile：<java.io.File> -
file_name：<文件名> -
correlationId： <UUID>（每行都一样） -
序列号：<n> -
序列大小：0（行数直到文件被读取时才知道）
有效载荷：
一个字符串每行。
第一行可以选择性地在前加一条带有开始标记有效载荷。
最后一行后面可选择性地跟一条带有结束标记有效载荷。
标记的存在和格式由以下带标记和markers-json性能。
模式 = 参考
头：
没有。
有效载荷：
一个java.io.file对象。
5.3.3. 选项
FTP源有以下选项：
按前缀分组的属性：
file.consumer
- markers-json
-
当“fileMarkers == true”时，指定它们应生成为FileSplitter.FileMarker对象还是JSON。（布尔值，默认：
true) - 模式
-
FileReadingMode 用于文件读取源。值包括“ref”——文件对象，“lines”——每行一条消息，或“contents”——内容以字节形式表示。（文件阅读模式，默认：
<没有>，可能的值：裁判,线,内容) - 带标记
-
设置为 true，以便在数据之前或之后发送文件开始/结束文件标记信息。仅适用于FileReadingMode的“行”。（布尔值，默认：
<没有>)
FTP.factory
- 缓存会话
-
缓存会话。（布尔值，默认：
<没有>) - 客户端模式
-
用于FTP会话的客户端模式。（客户端模式，默认：
<没有>，可能的值：积极,被动) - 主机
-
服务器的主机名。（字符串，默认：
本地主持) - 密码
-
连接服务器的密码。（字符串，默认：
<没有>) - 端口
-
服务器的移植。（整数，默认：
21) - 用户名
-
连接服务器的用户名。（字符串，默认：
<没有>)
FTP.Supplier
- 自动-创建-本地-dir
-
设置为true以创建本地目录（如果不存在）。（布尔值，默认：
true) - 延迟-空时
-
当检测到无新文件时的延迟持续时间。（持续时间，默认：
1) - 删除远程文件
-
设置为true以在成功传输后删除远程文件。（布尔值，默认：
false) - 文件名模式
-
一个过滤模式，用来匹配要传输的文件名称。（字符串，默认：
<没有>) - 文件名正则表达式
-
一个过滤正则表达式模式，用于匹配要传输的文件名称。（模式，默认：
<没有>) - 本地导引
-
用于文件传输的本地目录。（文件，默认：
<没有>) - 保留时间戳
-
设置为true以保留原始时间戳。（布尔值，默认：
true) - 远程指令
-
远程FTP目录。（字符串，默认：
/) - 远程文件分隔器
-
远程文件分离器。（字符串，默认：
/) - tmp-file-后缀
-
转移过程中使用的后缀。（字符串，默认：
.tmp)
Metadata.store.dynamo-db
- 创建延迟
-
创建表格重试之间的延迟。（整数，默认：
1) - 创建重试
-
创建表格请求的重试编号。（整数，默认：
25) - 读取容量
-
桌面上的读取容量。（长，默认：
1) - 桌子
-
元数据的表名。（字符串，默认：
<没有>) - 生存时间
-
表格条目的时间线。（整数，默认：
<没有>) - 写容量
-
写入容量放在桌面上。（长，默认：
1)
metadata.store.jdbc
- 地区
-
该存储库保留了消息的唯一分组标识符。（字符串，默认：
默认值) - 表前缀
-
自定义表名称的前缀。（字符串，默认：
<没有>)
Metadata.store.mongo-db
- 收集
-
MongoDB元数据集合名称。（字符串，默认：
元数据存储)
metadata.store.redis
- 钥匙
-
Redis 密钥用于元数据。（字符串，默认：
<没有>)
metadata.store
- 类型
-
表示要配置的元数据存储类型（默认为“内存”）。使用持久存储必须包含相应的 Spring Integration 依赖。（StoreType，默认：
<没有>，可能的值：蒙戈德,Redis,迪纳莫德,JDBC,Zookeeper,Hazelcast,存储)
元数据.store.zookeeper
- 连接串
-
Zookeeper 连接字符串格式为 HOST：PORT。（字符串，默认：
127.0.0.1:2181) - 编码
-
用于在Zookeeper中存储数据时使用的编码。（字符集，默认：
UTF-8) - 重试区间
-
Zookeeper作的重试间隔以毫秒计。（整数，默认：
1000) - 根
-
根节点——存储条目是该节点的子节点。（字符串，默认：
/SpringIntegration-元数据存储)
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请求并将正文作为消息载荷发出。
如果内容类型匹配文本/*或application/json，有效载荷将是字符串，
否则，有效载荷将是一个字节数组。
5.4.1. 有效载荷：
如果内容类型匹配文本/*或application/json
-
字符串
如果内容类型不匹配文本/*或application/json
-
字节数组
5.4.2. 选项
http 源代码支持以下配置属性：
按前缀分组的属性：
http.cors
- 允许凭证
-
浏览器是否应包含与被注释请求域名相关的任何Cookie。（布尔值，默认：
<没有>) - 允许的头部
-
在实际请求过程中可以使用的请求头列表。（字符串[]，默认：
<没有>) - 允许的起源
-
允许的起源列表，例如 https://domain1.com。（字符串[]，默认：
<没有>)
http
- 映射请求头
-
这些头部将被映射。（字符串[]，默认：
<没有>) - 路径模式
-
HTTP端点路径映射。（字符串，默认：
/)
服务器
- 端口
-
服务器HTTP端口。（整数，默认：
8080)
5.5. JDBC来源
该源从RDBMS轮询数据。
本资料完全基于DataSourceAutoConfiguration，更多信息请参见 Spring Boot JDBC 支持。
5.5.1. 有效载荷
-
Map<String，对象>什么时候jdbc.split == 真（默认） 和List<Map<String， Object>>否则
5.5.2. 选项
jdbc 源代码有以下选项：
按前缀分组的属性：
JDBC.提供商
- 最大行数
-
查询时处理的最大行数。（整数，默认：
0) - 查询
-
用于选择数据的查询。（字符串，默认：
<没有>) - 分裂
-
是否将SQL结果拆分为单独的消息。（布尔值，默认：
true) - 更新
-
一个用于标记轮询消息为“已见”的SQL更新语句。（字符串，默认：
<没有>)
spring.datasource
- 驾驶员类别名称
-
JDBCDrivers的完全合格姓名。默认根据网址自动检测。（字符串，默认：
<没有>) - 密码
-
数据库的登录密码。（字符串，默认：
<没有>) - 网址
-
数据库的JDBC网址。（字符串，默认：
<没有>) - 用户名
-
数据库用户名登录。（字符串，默认：
<没有>)
spring.integration.poller
- 克朗
-
Cron表达式用于投票。与“fixedDelay”和“fixedRate”互斥。（字符串，默认：
<没有>) - 固定延迟
-
投票延迟期。与“cron”和“fixedRate”互斥。（持续时间，默认：
<没有>) - 固定费率
-
投票率阶段。与“fixedDelay”和“cron”互斥。（持续时间，默认：
<没有>) - 初始延迟
-
初步延迟投票。申请了“固定延迟”和“固定费率”;被忽略，只用“克隆”。（持续时间，默认：
<没有>) - 每轮询最大消息数
-
每个轮询周期可轮询的最大消息数。（整数，默认：
<没有>) - 接收超时
-
投票通知要等多久？（持续时间，默认：
1)
另请参见 Spring Boot 文档以获取补充内容数据来源性质和触发属性和MaxMessagesProperties（最大信息属性）用于投票选项。
5.6. JMS源
JMS 源能够接收来自 JMS 的消息。
5.6.1. 选项
JMS 源代码有以下选项：
按前缀分组的属性：
JMS.提供商
- 客户端ID
-
持久订阅的客户ID。（字符串，默认：
<没有>) - 目的地
-
接收消息的目的地（队列或主题）。（字符串，默认：
<没有>) - 消息选择器
-
一个信息选择器。（字符串，默认：
<没有>) - 会话交易
-
True用于启用事务并选择DefaultMessageListenerContainer，false用于选择SimpleMessageListenerContainer。（布尔值，默认：
true) - 订阅-持久
-
对于耐用的订阅来说，这确实如此。（布尔值，默认：
<没有>) - 订阅名称
-
持久或共享订阅的名称。（字符串，默认：
<没有>) - 订阅共享
-
共享订阅确实如此。（布尔值，默认：
<没有>)
spring.jms
- JNDI名称
-
连接工厂JNDI名称。设置后，优先于其他连接的工厂自动配置。（字符串，默认：
<没有>) - 公文子域名
-
默认目的地类型是否为主题。（布尔值，默认：
false)
spring.jms.listener
- 确认模式
-
确认容器的模式。默认情况下，听者会被自动确认。（确认模式，默认：
<没有>，可能的值：自动,客户端,DUPS_OK) - 自动启动
-
启动时自动启动容器。（布尔值，默认：
true) - 并发
-
同时用户数量的最低要求。如果未指定最大并发，最小值也会被用作最大值。（整数，默认：
<没有>) - 最大并发
-
最大同时用户数量。（整数，默认：
<没有>) - 接收超时
-
超时用于接听电话。使用-1表示无等待接收，0表示完全无超时。后者仅在不在事务管理器中运行时可行，且通常不建议，因为它会防止干净的关闭。（持续时间，默认：
1)
5.7. 阿帕奇卡夫卡源
该模块接收Apache Kafka的消息。
5.7.1. 期权
卡夫卡资料有以下选项：
（请参阅 Spring Boot 文档中关于 Spring for Apache Kafka 配置属性的相关内容）
按前缀分组的属性：
Kafka.提供商
- ack-discarded
-
是否在“RecordFilterStrategy”之后确认已丢弃的记录。（布尔值，默认：
false) - 记录Filter
-
带有“ConsumerRecord”作为根对象的“RecordFilterStrategy”的SpEL表达式。（表达式，默认：
<没有>) - 主题模式
-
Apache Kafka 主题模式订阅。（模式，默认：
<没有>) - 主题
-
订阅Apache Kafka主题。（字符串[]，默认：
<没有>)
Spring。卡夫卡
- 自举服务器
-
用于建立与Kafka集群初始连接的host：port对列表。除非被覆盖，否则适用于所有组件。（List<String>，默认：
<没有>) - 客户端ID
-
发送请求时传递给服务器的ID。用于服务器端日志记录。（字符串，默认：
<没有>) - 性能
-
用于配置客户端的附加属性，这些属性是生产者和消费者共同的。（Map<String， String>，默认：
<没有>)
Spring.kafka.consumer
- 自动提交区间
-
如果将“enable.auto.commit”设置为true，消费者偏移量被自动提交到Kafka的频率。（持续时间，默认：
<没有>) - 自动偏移重置
-
当Kafka中没有初始偏移，或者当前偏移在服务器上已经不存在时该怎么办？（字符串，默认：
<没有>) - 自举服务器
-
用于建立与Kafka集群初始连接的host：port对列表。对消费者来说，覆盖了全球属性。（List<String>，默认：
<没有>) - 客户端ID
-
发送请求时传递给服务器的ID。用于服务器端日志记录。（字符串，默认：
<没有>) - enable-auto-commit
-
消费者的抵消是否在后台定期提交。（布尔值，默认：
<没有>) - 取-最大-等待
-
如果数据不足以立即满足“fetch-min-size”的要求，服务器在响应取指请求前，会在响应取用请求前分段的最大时间。（持续时间，默认：
<没有>) - 取最小大小
-
服务器应返回的最低数据量，用于获取请求。（DataSize，默认：
<没有>) - 群体识别
-
唯一字符串，标识该消费者所属的消费者组。（字符串，默认：
<没有>) - 心跳间隔
-
心跳之间的预期时间与消费者协调员。（持续时间，默认：
<没有>) - 隔离层级
-
用于读取事务式写入的消息的隔离层。（隔离级别，默认：
已读-未承诺，可能的值：READ_UNCOMMITTED,READ_COMMITTED) - 密钥解串器
-
密钥的解串器类。（职业<？>，默认：
<没有>) - 最大投票记录
-
单次调用 poll（ 中返回的最大记录数）。（整数，默认：
<没有>) - 性能
-
用于配置客户端的额外消费者特定属性。（Map<String， String>，默认：
<没有>) - 值解串器
-
用于值的反串行器类。（职业<？>，默认：
<没有>)
Spring.kafka.listener
- ack计数
-
当ackMode为“COUNT”或“COUNT_TIME”时，偏移提交之间的记录数。（整数，默认：
<没有>) - ack-mode
-
听众 AckMode。请参阅春季卡夫卡文档。（AckMode，默认：
<没有>，可能的值：记录,Batch,时间,计数,COUNT_TIME,手动,MANUAL_IMMEDIATE) - 降速时间
-
当ackMode为“TIME”或“COUNT_TIME”时，偏移提交间隔时间。（持续时间，默认：
<没有>) - 异步加速
-
支持异步记录确认。仅适用于 spring.kafka.listener.ack-mode 是手动或手动即时模式。（布尔值，默认：
<没有>) - 自动启动
-
是否自动启动集装箱。（布尔值，默认：
true) - 客户端ID
-
听者的消费者 client.id 属性前缀。（字符串，默认：
<没有>) - 并发
-
监听器容器中要运行的线程数量。（整数，默认：
<没有>) - 闲置间民调
-
Consumer.poll（Duration）调用之间的睡眠间隔。（持续时间，默认：
0) - 空闲事件区间
-
闲置消费者事件发布间隔时间（未接收数据）。（持续时间，默认：
<没有>) - 空闲-划分-事件-区间
-
发布空闲分区消费者事件之间的时间（分区未收到数据传输）。（持续时间，默认：
<没有>) - 即停
-
容器是在当前记录处理完毕后停止，还是在处理完上一次轮询的所有记录后才停止。（布尔值，默认：
false) - log-container-config
-
是否在初始化（INFO级别）记录容器配置。（布尔值，默认：
<没有>) - 缺失主题-致命
-
如果中介中至少有一个配置的主题不存在，容器是否应该启动失败。（布尔值，默认：
false) - 监听音程
-
对于不响应的消费者，检查间隔时间。如果未指定持续时间后缀，则使用秒。（持续时间，默认：
<没有>) - 无轮询阈值
-
乘数应用到“pollTimeout”以判断消费者是否无响应。（浮点，默认：
<没有>) - 投票超时
-
在投票消费者时使用超时。（持续时间，默认：
<没有>) - 类型
-
倾听型。（类型，默认：
单)
5.8. 负载发生源
一个发送生成数据并分发到流的源。
5.8.1. 期权
负载发生源有以下选项：
- load-generator.generate-timestamp
-
时间戳是否生成。（布尔值，默认：
false) - load-generator.message-count
-
消息数量。（整数，默认：
1000) - load-generator.message-size
-
消息大小。（整数，默认：
1000) - 负载生成器。生产者
-
生产者的数量。（整数，默认：
1)
5.9. 邮件源
一个源应用程序，监听邮件并将消息正文作为消息负载发出。
5.9.1. 期权
邮件源有以下选项：
- mail.supplier.charset
-
字节[]邮件转字符串转换的字符集。（字符串，默认：
UTF-8) - mail.supplier.delete
-
设置为true以删除下载后的邮件。（布尔值，默认：
false) - Mail.supplier.expression
-
配置一个 SpEL 表达式来选择消息。（字符串，默认：
true) - mail.supplier.idle-imap
-
设置为true以使用IdleImap配置。（布尔值，默认：
false) - mail.supplier.java-mail-properties（-mail-properties）
-
JavaMail 属性作为新行分隔的名称-值对字符串，例如 'foo=bar\n baz=car'。（属性，默认：
<没有>) - Mail.supplier.标记为已读
-
设置为true以标记邮件为已读。（布尔值，默认：
false) - Mail.supplier.url
-
邮件连接链接，用于连接邮件服务器，例如“imaps://username：[email protected]：993/Inbox'。（URLname，默认：
<没有>) - Mail.supplier.user-flag
-
当服务器不支持\Recent。（字符串，默认：
<没有>)
5.10. MongoDB 源
该来源来自MongoDB的数据轮询。
本资料完全基于MongoDataAutoConfiguration，更多信息请参阅 Spring Boot MongoDB 支持。
5.10.1. 选项
mongodb 源代码有以下选项：
按前缀分组的属性：
Mongodb.supplier
- 收集
-
查询MongoDB集合。（字符串，默认：
<没有>) - 查询
-
MongoDB查询。（字符串，默认：
{ }) - 查询表达式
-
MongoDB 查询中的 SpEL 表达式为 DSL 格式。（表达式，默认：
<没有>) - 分裂
-
是否将查询结果拆分为单独的消息。（布尔值，默认：
true) - 更新表达式
-
MongoDB 中的 SpEL 表达式是更新 DSL 风格的。（表达式，默认：
<没有>)
spring.data.mongodb
- 额外主持人
-
额外的服务器主机。如果没有指定“host”，则无法用URI设置。其他主机将使用默认的Mongo端口27017。如果你想用不同的端口，可以用“host：port”语法。（List<String>，默认：
<没有>) - 认证数据库
-
认证数据库名称。（字符串，默认：
<没有>) - 自动索引创建
-
是否启用自动索引创建。（布尔值，默认：
<没有>) - 数据库
-
数据库名称。覆盖URI数据库。（字符串，默认：
<没有>) - 场命名策略
-
字段命名策略的完全限定名称。（职业<？>，默认：
<没有>) - 主机
-
Mongo服务器主机。无法用URI设置。（字符串，默认：
<没有>) - 密码
-
Mongo服务器的登录密码。无法用URI设置。（字符[]，默认：
<没有>) - 端口
-
Mongo服务器端口。无法用URI设置。（整数，默认：
<没有>) - 复制集名称
-
集群的副本集名称要求。无法用URI设置。（字符串，默认：
<没有>) - 乌里
-
Mongo数据库URI。覆盖主机、端口、用户名和密码。（字符串，默认：
mongodb://localhost/test) - 用户名
-
登录Mongo服务器的用户。无法用URI设置。（字符串，默认：
<没有>) - UUID 表示
-
将 UUID 转换为 BSON 二进制值时使用的表示方式。（UuidRepresentation，默认：
Java-遗产，可能的值：未具体说明,标准,C_SHARP_LEGACY,JAVA_LEGACY,PYTHON_LEGACY)
spring.data.mongodb.gridfs
- 桶
-
GridFS 桶名称。（字符串，默认：
<没有>) - 数据库
-
GridFS 数据库名称。（字符串，默认：
<没有>)
spring.data.mongodb.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<没有>) - 启用
-
是否启用SSL支持。除非另有说明，否则若提供了“捆绑”，则自动启用。（布尔值，默认：
<没有>)
更多内容请参见 Spring Boot 文档MongoProperties性能。
参见 和触发属性用于投票选项。
5.11. MQTT 来源
能够接收来自MQTT消息的源。
5.11.1. 有效载荷：
-
字符串如果二元设置为false（默认） -
字节[]如果二元设置为true
5.11.2. 期权
mqtt 源有以下选项：
按前缀分组的属性：
MQTT
- 清洁录音
-
客户端和服务器是否应该在重启和重新连接时记住状态。（布尔值，默认：
true) - 连接超时
-
连接超时只需几秒钟。（整数，默认：
30) - 保持生命间隔
-
延迟以秒计。（整数，默认：
60) - 密码
-
连接经纪人时使用的密码。（字符串，默认：
客人) - 坚持
-
“记忆”或“文件”。（字符串，默认：
存储) - 持久性目录
-
持久化目录。（字符串，默认：
/tmp/帕霍) - SSL-性质
-
MQTT 客户端 SSL 属性。（Map<String， String>，默认：
<没有>) - 网址
-
MQTT经纪人的位置（逗号分隔列表）。（字符串[]，默认：
[tcp://localhost:1883]) - 用户名
-
连接经纪人时使用的用户名。（字符串，默认：
客人)
MQTT.提供商
- 二元的
-
true，将有效载荷保持为字节。（布尔值，默认：
false) - 字符集
-
用于将字节转换为字符串的字元集（当二进制为假时）。（字符串，默认：
UTF-8) - 客户端ID
-
识别客户。（字符串，默认：
stream.client.id.source) - QoS（服务质量）
-
QoS;所有主题都用一个单一值，或者用逗号分隔的列表来匹配主题。（整数[]，默认：
[0]) - 主题
-
来源将订阅的主题（逗号分隔）。（字符串[]，默认：
[stream.mqtt])
5.12. RabbitMQ 来源
“rabbit”源能够接收来自RabbitMQ的消息。
队列必须在流部署前存在;它们不是自动生成的。 你可以用RabbitMQ的网页界面轻松创建队列。
5.12.1. 输入
无
5.12.2. 输出
有效载荷
-
字节[]
5.12.3. 选项
兔子资源有以下选项：
按前缀分组的属性：
rabbit.supplier
- 使得重试
-
确实可以启用重试。（布尔值，默认：
false) - 初始重试区间
-
启用重试时的初始重试间隔。（整数，默认：
1000) - 映射请求头
-
这些头部将被映射。（字符串[]，默认：
[STANDARD_REQUEST_HEADERS]) - 最大尝试次数
-
启用重试时，最大投递次数。（整数，默认：
3) - 最大重试区间
-
当重试被启用时，重试间隔最大。（整数，默认：
30000) - 自有连接
-
如果是真的，根据启动属性使用单独的连接。（布尔值，默认：
false) - 队列
-
源端会监听消息的队列。（字符串[]，默认：
<没有>) - 重新排队
-
拒绝消息是否应重新排队。（布尔值，默认：
true) - 重试乘法
-
当重试开启时，重试回退倍率。（双重，默认：
2) - 交易
-
通道是否被交易。（布尔值，默认：
false)
Spring.rabbitmq
- 地址洗牌模式
-
用于洗牌配置地址的模式。（AddressShuffleMode，默认：
没有，可能的值：没有,随机,顺序) - 地址
-
客户端应连接的地址列表，以逗号分隔。设置好后，主机和端口被忽略。（字符串，默认：
<没有>) - 通道-RPC超时
-
通道中RPC呼叫的续续超时。把它设为零，等待很久。（持续时间，默认：
10米) - 连接超时
-
连接超时。把它设为零，等待很久。（持续时间，默认：
<没有>) - 主机
-
RabbitMQ主机。如果地址被设置了，则忽略。（字符串，默认：
本地主持) - 密码
-
登录以验证经纪人身份。（字符串，默认：
客人) - 端口
-
RabbitMQ移植版。如果地址被设置了，则忽略。默认使用5672，启用SSL则使用5671。（整数，默认：
<没有>) - 出版者-确认-类型
-
出版商类型确认使用。（ConfirmType，默认：
<没有>，可能的值：简单,相关,没有) - 出版商回归
-
是否启用出版商退货。（布尔值，默认：
false) - 请求通道最大值
-
客户端请求的每个连接的频道数量。无限用0。（整数，默认：
2047) - 请求心跳
-
请求心跳暂停;零对零。如果未指定持续时间后缀，则使用秒。（持续时间，默认：
<没有>) - 用户名
-
登录用户以向经纪人进行身份验证。（字符串，默认：
客人) - 虚拟主机
-
连接经纪商时使用的虚拟主机。（字符串，默认：
<没有>)
spring.rabbitmq.listener.simple
- 确认模式
-
确认容器模式。（确认模式，默认：
<没有>，可能的值：没有,手动,自动) - 自动启动
-
是否在启动时自动启动容器。（布尔值，默认：
true) - 批次大小
-
批处理大小，表示容器将使用物理消息的数量。（整数，默认：
<没有>) - 并发
-
最小数量的监听调用线程。（整数，默认：
<没有>) - 消费者批处理启用
-
容器是否基于“接收超时”和“批次大小”创建一批消息。强制将“de-batching-enabled”设置为true，将生产者创建批次的内容作为离散记录包含在该批次中。（布尔值，默认：
false) - 启用批次处理
-
容器是否应该以离散消息的形式呈现，还是用批次呼叫监听者。（布尔值，默认：
true) - 默认-重新排队-拒绝
-
是否默认会重新排队被拒绝的投递。（布尔值，默认：
<没有>) - 空闲事件区间
-
闲置容器事件应多久发布一次。（持续时间，默认：
<没有>) - 最大并发
-
最大监听调用线程数。（整数，默认：
<没有>) - 缺失队列-致命
-
如果容器声明的队列在代理中不可用，是否失败;或者在运行时删除一个或多个队列时是否停止容器。（布尔值，默认：
true) - 预取
-
每个消费者最多可未收到的未确认消息数量。（整数，默认：
<没有>)
Spring.兔子 mq.listener
- 类型
-
监听器容器类型。（ContainerType，默认：
简单，可能的值：简单,直接,流)
另请参阅 Spring Boot 文档，了解代理连接和监听者属性的附加属性。
关于重试的说明
使用默认的ackMode（AUTO）和requeue（true）选项，失败的消息传递将被重新尝试
无限期。
由于兔源处理较少，源本身故障的风险较小，除非
下游粘结 剂不知为何没有连接。
将 requeue 设置为 false 会导致消息在第一次尝试时被拒绝（甚至可能发送到死信
如果经纪人配置了交换/队列）。
enableRetry 选项允许配置重试参数，使得失败的消息传递可以重新尝试，
当重试次数用尽时，最终会被丢弃（或写死字母）。
在重试期间，传递线程会被暂停。
重试选项有 enableRetry、maxAttempts、initialRetryInterval、retryMultiplier 和 maxRetryInterval。
带有MessageConversionException的消息传递失败时，绝不会重试;假设如果
第一次尝试无法转换，后续尝试也会失败。
此类消息会被丢弃（或死字母处理）。 |
5.12.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps你可以在这里找到相应的活页夹项目。 然后你可以把光盘放进某个文件夹里，然后构建它：
$ ./mvnw clean package5.12.5. 示例
java -jar rabbit-source.jar --rabbit.queues=5.13. 亚马逊S3来源
该源应用支持使用亚马逊S3协议传输文件。
文件是从远程目录（S3桶）映射到当地应用程序部署的目录。
源端发出的消息默认以字节数组形式提供。然而，这可以是
通过--模式选择：
-
裁判提供
java.io.file参考 -
线会逐行拆分文件，并为每行发送一条新消息
-
内容默认。以字节数组的形式提供文件内容
使用--模式=线你也可以提供额外的选项--withMarkers=真.
如果设置为true， 基础文件分拆器会在实际数据之前和之后发送额外的文件开头和结束标记信息。
这两个额外标记消息的有效载荷为FileSplitter.FileMarker.选项withMarkers默认为false如果没有明确设置。
另见元数据存储选项，了解可能的共享持久存储配置，以防止重启时出现重复消息。
5.13.1. 模式 = 线
头：
-
内容类型：文本/纯文字 -
file_orginalFile：<java.io.File> -
file_name：<文件名> -
correlationId： <UUID>（每行都一样） -
序列号：<n> -
序列大小：0（行数直到文件被读取时才知道）
有效载荷：
一个字符串每行。
第一行可以选择性地在前加一条带有开始标记有效载荷。
最后一行后面可选择性地跟一条带有结束标记有效载荷。
标记的存在和格式由以下带标记和markers-json性能。
5.13.2. mode = 参考
头：
没有。
有效载荷：
一个java.io.file对象。
5.13.3. 选项
这s3-源有以下选项：
按前缀分组的属性：
file.consumer
- markers-json
-
当“fileMarkers == true”时，指定它们应生成为FileSplitter.FileMarker对象还是JSON。（布尔值，默认：
true) - 模式
-
FileReadingMode 用于文件读取源。值包括“ref”——文件对象，“lines”——每行一条消息，或“contents”——内容以字节形式表示。（文件阅读模式，默认：
<没有>，可能的值：裁判,线,内容) - 带标记
-
设置为 true，以便在数据之前或之后发送文件开始/结束文件标记信息。仅适用于FileReadingMode的“行”。（布尔值，默认：
<没有>)
Metadata.store.dynamo-db
- 创建延迟
-
创建表格重试之间的延迟。（整数，默认：
1) - 创建重试
-
创建表格请求的重试编号。（整数，默认：
25) - 读取容量
-
桌面上的读取容量。（长，默认：
1) - 桌子
-
元数据的表名。（字符串，默认：
<没有>) - 生存时间
-
表格条目的时间线。（整数，默认：
<没有>) - 写容量
-
写入容量放在桌面上。（长，默认：
1)
metadata.store.jdbc
- 地区
-
该存储库保留了消息的唯一分组标识符。（字符串，默认：
默认值) - 表前缀
-
自定义表名称的前缀。（字符串，默认：
<没有>)
Metadata.store.mongo-db
- 收集
-
MongoDB元数据集合名称。（字符串，默认：
元数据存储)
metadata.store.redis
- 钥匙
-
Redis 密钥用于元数据。（字符串，默认：
<没有>)
metadata.store
- 类型
-
表示要配置的元数据存储类型（默认为“内存”）。使用持久存储必须包含相应的 Spring Integration 依赖。（StoreType，默认：
<没有>，可能的值：蒙戈德,Redis,迪纳莫德,JDBC,Zookeeper,Hazelcast,存储)
元数据.store.zookeeper
- 连接串
-
Zookeeper 连接字符串格式为 HOST：PORT。（字符串，默认：
127.0.0.1:2181) - 编码
-
用于在Zookeeper中存储数据时使用的编码。（字符集，默认：
UTF-8) - 重试区间
-
Zookeeper作的重试间隔以毫秒计。（整数，默认：
1000) - 根
-
根节点——存储条目是该节点的子节点。（字符串，默认：
/SpringIntegration-元数据存储)
S3.提供商
- 自动-创建-本地-dir
-
创建或不创建本地目录。（布尔值，默认：
true) - 删除远程文件
-
处理完后删除或不删除远程文件。（布尔值，默认：
false) - 文件名模式
-
过滤远程文件的模式。（字符串，默认：
<没有>) - 文件名正则表达式
-
正规表达式用于过滤远程文件。（模式，默认：
<没有>) - 仅列表
-
设置为 true，以返回 s3 对象元数据，但不复制文件到本地目录。（布尔值，默认：
false) - 本地导引
-
本地目录用于存储文件。（文件，默认：
<没有>) - 保留时间戳
-
是否将远程文件的时间戳转移到本地文件。（布尔值，默认：
true) - 远程指令
-
AWS S3 桶资源。（字符串，默认：
桶) - 远程文件分隔器
-
远程文件分离器。（字符串，默认：
/) - tmp-file-后缀
-
临时文件后缀。（字符串，默认：
.tmp)
Spring.cloud.aws.credentials
- 访问密钥
-
访问密钥用于静态提供者。（字符串，默认：
<没有>) - 实例配置文件
-
无需额外配置即可配置实例配置文件凭证提供者。（布尔值，默认：
false) - 轮廓
-
AWS配置文件。（配置文件，默认：
<没有>) - 秘密密钥
-
与静态提供者一起使用的密钥。（字符串，默认：
<没有>)
spring.cloud.aws.region
- 实例配置文件
-
无需额外配置即可配置实例配置文件区域提供者。（布尔值，默认：
false) - 轮廓
-
AWS配置文件。（配置文件，默认：
<没有>) - 静态的
-
<缺少文档>（字符串，默认：
<没有>)
春云.AWS3
- 加速模式启用
-
在访问 S3 时启用加速端点的选项。加速端点通过利用亚马逊云前线的全球分布式边缘位置，实现了更快速的对象传输。（布尔值，默认：
<没有>) - 校验和验证启用
-
有个选项可以禁用对存储在 S3 中对象的校验和进行验证。（布尔值，默认：
<没有>) - 分块编码支持功能
-
在签署请求载荷时启用分块编码的选项，适用于{@link software.amazon.awssdk.services.s3.model.PutObjectRequest}和{@link software.amazon.awssdk.services.s3.model.UploadPartRequest}。（布尔值，默认：
<没有>) - 跨区域支持
-
支持跨区域桶访问。（布尔值，默认：
<没有>) - 端点
-
覆盖默认端点。（URI，默认：
<没有>) - 路径-样式-访问-启用
-
启用路径风格访问以访问S3对象，而非DNS风格访问的选项。偏好DNS风格的访问，因为这样在访问S3时能实现更好的负载均衡。（布尔值，默认：
<没有>) - 地区
-
覆盖默认区域。（字符串，默认：
<没有>) - use-arn-region-enabled
-
如果作为S3作的目标传递了一个与客户端配置区域不同的S3作目标，则该标志必须设置为“true”，以便客户端可以对ARN指定区域进行跨区域调用，否则将抛出异常。（布尔值，默认：
<没有>)
5.13.4. 亚马逊AWS常见选项
Amazon S3 Source（和所有其他 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协议传输文件。
文件是从远程目录当地应用部署的目录。
源端发出的消息默认以字节数组形式提供。然而，这可以是
通过--模式选择：
-
裁判提供
java.io.file参考 -
线会逐行拆分文件，并为每行发送一条新消息
-
内容默认。以字节数组的形式提供文件内容
使用--模式=线你也可以提供额外的选项--withMarkers=真.
如果设置为true， 基础文件分拆器会在实际数据之前和之后发送额外的文件开头和结束标记信息。
这两个额外标记消息的有效载荷为FileSplitter.FileMarker.选项withMarkers默认为false如果没有明确设置。
看SFTP-提供商用于高级配置选项。
另见元数据存储选项，了解可能的共享持久存储配置，以防止重启时出现重复消息。
5.14.1. 输入
N/A（从SFTP服务器获取文件）。
5.14.2. 输出
模式 = 目录
头：
-
内容类型：application/octet-stream -
file_name：<文件名> -
file_remoteFileInfo <文件元数据> -
file_remoteHostPort：<host：port> -
file_remoteDirectory：<相对路径> -
file_remoteFile：<文件名> -
sftp_selectedServer：<服务器密钥>（如果是多源）
有效载荷：
一个字节[]装满了文件内容。
模式 = 线
头：
-
内容类型：文本/纯文字 -
file_name：<文件名> -
correlationId： <UUID>（每行都一样） -
序列号：<n> -
序列大小：0（行数直到文件被读取时才知道） -
file_marker ：<文件标记>（如果启用了带标记）
有效载荷：
一个字符串每行。
第一行可以选择性地在前加一条带有开始标记有效载荷。
最后一行后面可选择性地跟一条带有结束标记有效载荷。
标记的存在和格式由以下带标记和markers-json性能。
模式 = 参考
头：
-
file_remoteHostPort：<host：port> -
file_remoteDirectory：<相对路径> -
file_remoteFile：<文件名> -
file_originalFile：<本地文件的绝对路径> -
file_name <本地文件名> -
file_relativePath -
file_remoteFile：<远程文件名> -
sftp_selectedServer：<服务器密钥>（如果是多源）
有效载荷：
一个java.io.file对象。
5.14.3. 选项
FTP源有以下选项：
按前缀分组的属性：
file.consumer
- markers-json
-
当“fileMarkers == true”时，指定它们应生成为FileSplitter.FileMarker对象还是JSON。（布尔值，默认：
true) - 模式
-
FileReadingMode 用于文件读取源。值包括“ref”——文件对象，“lines”——每行一条消息，或“contents”——内容以字节形式表示。（文件阅读模式，默认：
<没有>，可能的值：裁判,线,内容) - 带标记
-
设置为 true，以便在数据之前或之后发送文件开始/结束文件标记信息。仅适用于FileReadingMode的“行”。（布尔值，默认：
<没有>)
Metadata.store.dynamo-db
- 创建延迟
-
创建表格重试之间的延迟。（整数，默认：
1) - 创建重试
-
创建表格请求的重试编号。（整数，默认：
25) - 读取容量
-
桌面上的读取容量。（长，默认：
1) - 桌子
-
元数据的表名。（字符串，默认：
<没有>) - 生存时间
-
表格条目的时间线。（整数，默认：
<没有>) - 写容量
-
写入容量放在桌面上。（长，默认：
1)
metadata.store.jdbc
- 地区
-
该存储库保留了消息的唯一分组标识符。（字符串，默认：
默认值) - 表前缀
-
自定义表名称的前缀。（字符串，默认：
<没有>)
Metadata.store.mongo-db
- 收集
-
MongoDB元数据集合名称。（字符串，默认：
元数据存储)
metadata.store.redis
- 钥匙
-
Redis 密钥用于元数据。（字符串，默认：
<没有>)
metadata.store
- 类型
-
表示要配置的元数据存储类型（默认为“内存”）。使用持久存储必须包含相应的 Spring Integration 依赖。（StoreType，默认：
<没有>，可能的值：蒙戈德,Redis,迪纳莫德,JDBC,Zookeeper,Hazelcast,存储)
元数据.store.zookeeper
- 连接串
-
Zookeeper 连接字符串格式为 HOST：PORT。（字符串，默认：
127.0.0.1:2181) - 编码
-
用于在Zookeeper中存储数据时使用的编码。（字符集，默认：
UTF-8) - 重试区间
-
Zookeeper作的重试间隔以毫秒计。（整数，默认：
1000) - 根
-
根节点——存储条目是该节点的子节点。（字符串，默认：
/SpringIntegration-元数据存储)
SFTP.提供商
- 自动-创建-本地-dir
-
设置为true以创建本地目录（如果不存在）。（布尔值，默认：
true) - 延迟-空时
-
当检测到无新文件时的延迟持续时间。（持续时间，默认：
1) - 删除远程文件
-
设置为true以在成功传输后删除远程文件。（布尔值，默认：
false) - 目录
-
一份工厂“name.directory”对的列表。（字符串[]，默认：
<没有>) - 工厂
-
工厂名称映射到工厂。（Map<String， Factory>，默认：
<没有>) - 公平
-
对于多个服务器/目录的公平轮换来说，确实如此。默认情况下，这个错误，因此如果一个来源有多个条目，这些条目会在访问其他来源之前被接收。（布尔值，默认：
false) - 文件名模式
-
一个过滤模式，用来匹配要传输的文件名称。（字符串，默认：
<没有>) - 文件名正则表达式
-
一个过滤正则表达式模式，用于匹配要传输的文件名称。（模式，默认：
<没有>) - 仅列表
-
设置为 true，以返回文件元数据而不包含全部有效载荷。（布尔值，默认：
false) - 本地导引
-
用于文件传输的本地目录。（文件，默认：
<没有>) - 最大取物
-
每次轮询可获取的最大远程文件数;默认无限。在列出文件或构建任务启动请求时不适用。（整数，默认：
<没有>) - 保留时间戳
-
设置为true以保留原始时间戳。（布尔值，默认：
true) - 远程指令
-
远程FTP目录。（字符串，默认：
/) - 远程文件分隔器
-
远程文件分离器。（字符串，默认：
/) - rename-remote-files-to
-
解析为新名称的 SpEL 表达式在成功传输后必须重命名为（表达式，默认：
<没有>) - 流
-
设置为 true，以便流式传输文件，而不是复制到本地目录。（布尔值，默认：
false) - tmp-file-后缀
-
转移过程中使用的后缀。（字符串，默认：
.tmp)
SFTP.提供商.工厂
- 允许未知密钥
-
允许未知或更改的密钥确实如此。（布尔值，默认：
false) - 主机
-
服务器的主机名。（字符串，默认：
本地主持) - 已知宿主表达式
-
一个 SpEL 表达式，解析为已知主机文件的位置。（表达式，默认：
<没有>) - 暗语
-
用户私钥的密码短语。（字符串，默认：
<空字符串>) - 密码
-
连接服务器的密码。（字符串，默认：
<没有>) - 端口
-
服务器的移植。（整数，默认：
22) - 私钥
-
用户私钥的资源位置。（资源，默认：
<没有>) - 用户名
-
连接服务器的用户名。（字符串，默认：
<没有>)
sftp.supplier.sort-by
- 属性
-
文件列表条目的属性按（FILENAME， ATIME： 最后访问时间， MTIME： 最后修改时间） 排序。（属性，默认：
<没有>) - 迪尔
-
排序方向（ASC或DESC）。（作者，默认：
<没有>)
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. 系统日志
syslog 源通过 UDP、TCP 或两者都接收 SYSLOG 数据包。支持RFC3164（BSD）和RFC5424格式。
5.15.1. 选项
- syslog.supplier.buffer-size
-
解码消息时使用的缓冲区大小;较大的消息将被拒绝。（整数，默认：
2048) - syslog.supplier.nio
-
是否使用NIO（支持大量连接）。（布尔值，默认：
false) - syslog.supplier.port
-
听的端口。（整数，默认：
1514) - syslog.supplier.protocol
-
用于SYSLOG的协议（TCP或UDP）。（协议，默认：
<没有>，可能的值：TCP,UDP,双) - syslog.supplier.reverse-lookup
-
是否对输入插槽进行反向查找。（布尔值，默认：
false) - syslog.supplier.rfc
-
“5424”或“3164”——根据RFC的syslog格式;3164 也被称为“BSD”格式。（字符串，默认：
3164) - syslog.supplier.socket-timeout
-
插槽的超时。（整数，默认：
0)
5.16. TCP
这TCP源代码作为服务器，允许远程方连接并通过原始TCP套接字提交数据。
TCP是一种流式传输协议，需要某种机制来在线路上对消息进行帧。有若干解码器 可用，默认为“CRLF”，兼容Telnet。
TCP源应用程序产生的消息具有字节[]有效载荷。
5.16.1. 期权
按前缀分组的属性：
TCP
- 蔚来
-
是否使用蔚来。（布尔值，默认：
false) - 端口
-
值得倾听的端口;作系统选择端口时，0。（整数，默认：
1234) - 反向查找
-
对远程IP地址进行反向DNS查找;如果为假，消息头部中仅包含IP地址。（布尔值，默认：
false) - 套接字超时
-
当未接收到数据时，关闭套接字前的超时（毫秒）。（整数，默认：
120000) - 使用直接缓冲区
-
是否使用直接缓冲。（布尔值，默认：
false)
TCP.提供商
- 缓冲区大小
-
解码消息时使用的缓冲区大小;较大的消息将被拒绝。（整数，默认：
2048) - 译码器
-
接收消息时使用的解码器。（编码，默认：
<没有>，可能的值：CRLF,如果,零,STXETX,生,第一语言,L2,L4)
5.16.2. 可用解码器
- CRLF（默认）
-
文本以回车（0x0d）结束，随后换行（0x0a）
- 如果
-
文本以换行符结束（0x0a）
- 零
-
以空字节（0x00）结束的文本
- STXETX
-
文本前置STX（0x02），结尾为ETX（0x03）
- 生
-
无结构——客户端通过关闭套接字表示完整消息
- 第一语言
-
数据前有一个一字节（无符号）长度字段（支持最多255字节）
- L2
-
数据前有一个两字节（无符号）长度字段（最多216-1字节）
- L4
-
数据前有一个四字节（带符号）长度字段（最多231-1字节）
5.17. 时间源
时间源会定期发出一个包含当前时间的字符串。
5.17.1. 期权
时间源有以下选项：
- spring.integration.poller.cron
-
Cron表达式用于投票。与“fixedDelay”和“fixedRate”互斥。（字符串，默认：
<没有>) - spring.integration.poller.fixed-delay
-
投票延迟期。与“cron”和“fixedRate”互斥。（持续时间，默认：
<没有>) - spring.integration.poller.fixed-rate
-
投票率阶段。与“fixedDelay”和“cron”互斥。（持续时间，默认：
<没有>) - spring.integration.poller.initial-delay
-
初步延迟投票。申请了“固定延迟”和“固定费率”;被忽略，只用“克隆”。（持续时间，默认：
<没有>) - 每轮询 spring.integration.poller.max 条消息
-
每个轮询周期可轮询的最大消息数。（整数，默认：
<没有>) - spring.integration.poller.receive-timeout
-
投票通知要等多久？（持续时间，默认：
1) - 时间.日期格式
-
日期值格式。（字符串，默认：
嗯/dd/yy HH：mm：ss。)
5.18. 推特消息来源
反复检索过去30天内的直接消息（包括发送和接收），按倒时间顺序排序。
解除的消息被缓存（在元数据存储缓存）以防止重复。
默认情况下，内存中SimpleMetadataStore被使用。
这推特.消息.来源.计数控制返回消息的数量。
这spring.cloud.stream.poller属性控制消息轮询间隔。
必须与已使用API的速率限制保持一致
5.18.1. 期权
按前缀分组的属性：
spring.integration.poller
- 克朗
-
Cron表达式用于投票。与“fixedDelay”和“fixedRate”互斥。（字符串，默认：
<没有>) - 固定延迟
-
投票延迟期。与“cron”和“fixedRate”互斥。（持续时间，默认：
<没有>) - 固定费率
-
投票率阶段。与“fixedDelay”和“cron”互斥。（持续时间，默认：
<没有>) - 初始延迟
-
初步延迟投票。申请了“固定延迟”和“固定费率”;被忽略，只用“克隆”。（持续时间，默认：
<没有>) - 每轮询最大消息数
-
每个轮询周期可轮询的最大消息数。（整数，默认：
<没有>) - 接收超时
-
投票通知要等多久？（持续时间，默认：
1)
Twitter.connection
- 访问Tokens
-
你的TwitterTokens。（字符串，默认：
<没有>) - 访问Tokens秘密
-
你的推特Tokens秘密。（字符串，默认：
<没有>) - 消费者密钥
-
你的推特密钥。（字符串，默认：
<没有>) - 消费者秘密
-
你的推特秘密。（字符串，默认：
<没有>) - 调试支持
-
启用 Twitter4J 调试模式。（布尔值，默认：
false) - raw-json
-
启用缓存由 Twitter API 返回的原始（原始）JSON 对象。当设置为 False 时，结果将使用 Twitter4J 的 json 表示。设置为 True 时，结果将使用原始的 Twitter APIS json 表示。（布尔值，默认：
true)
推特.信息.来源
- 计数
-
最多要返回的事件数量。20 默认。最多50。（整数，默认：
20)
5.19. 推特搜索源
Twitter的标准搜索API（搜索/推文）允许对最近或热门推文的索引进行简单查询。这源对过去7天内发布的推文进行持续搜索。属于“公开”API集。
返回一组匹配特定查询的相关推文。
使用该spring.cloud.stream.poller属性用于控制连续搜索请求之间的间隔。速率限制 - 每30分钟窗口180个请求（例如~6转/分钟，~1个请求/10秒）
这推特.搜索查询属性允许按关键词查询，并按时间和地理位置筛选结果。
这推特.search.count和twitter.search.page根据搜索API控制结果分页。
注意：Twitter的搜索服务及其延伸的搜索API并非推文的详尽来源。并非所有推文都会被索引或通过搜索界面提供。
5.19.1. 期权
按前缀分组的属性：
spring.integration.poller
- 克朗
-
Cron表达式用于投票。与“fixedDelay”和“fixedRate”互斥。（字符串，默认：
<没有>) - 固定延迟
-
投票延迟期。与“cron”和“fixedRate”互斥。（持续时间，默认：
<没有>) - 固定费率
-
投票率阶段。与“fixedDelay”和“cron”互斥。（持续时间，默认：
<没有>) - 初始延迟
-
初步延迟投票。申请了“固定延迟”和“固定费率”;被忽略，只用“克隆”。（持续时间，默认：
<没有>) - 每轮询最大消息数
-
每个轮询周期可轮询的最大消息数。（整数，默认：
<没有>) - 接收超时
-
投票通知要等多久？（持续时间，默认：
1)
Twitter.connection
- 访问Tokens
-
你的TwitterTokens。（字符串，默认：
<没有>) - 访问Tokens秘密
-
你的推特Tokens秘密。（字符串，默认：
<没有>) - 消费者密钥
-
你的推特密钥。（字符串，默认：
<没有>) - 消费者秘密
-
你的推特秘密。（字符串，默认：
<没有>) - 调试支持
-
启用 Twitter4J 调试模式。（布尔值，默认：
false) - raw-json
-
启用缓存由 Twitter API 返回的原始（原始）JSON 对象。当设置为 False 时，结果将使用 Twitter4J 的 json 表示。设置为 True 时，结果将使用原始的 Twitter APIS json 表示。（布尔值，默认：
true)
推特.搜索
- 计数
-
每页（例如每个单次请求）可返回的推文数量，最多可达100条。（整数，默认：
100) - 朗
-
限制搜索推文只能使用由 http://en.wikipedia.org/wiki/ISO_639-1 给定的语言。（字符串，默认：
<没有>) - 页
-
在重新从最近一条推文开始搜索之前，需要倒查页面数（例如请求数）。倒推推文总数为（页数乘以）（整数，默认：
3) - 查询
-
通过搜索查询词搜索推文。（字符串，默认：
<没有>) - 从最近最多回复开始
-
从最近的推文开始搜索，回复空。仅在第一次重启后应用（例如since_id ！= 无界）（布尔值，默认：
false) - 结果类型
-
具体说明你希望获得哪种类型的搜索结果。目前默认状态是“混合型”。有效值包括：混合：在回答中包含流行和实时结果。最近：仅返回回答中最新的结果 热门：返回回答中最受欢迎的结果（ResultType，默认：
<没有>，可能的值：流行,混合,最近) - 因为
-
如有指定，回传自指定日期以来的推文。日期应格式为 YYYY-MM-DD。（字符串，默认：
<没有>)
推特.search.geocode
- 纬度
-
用户自由度。（双重，默认：
-1) - 经度
-
用户的经度。（双重，默认：
-1) - 半径
-
以（纬度、经度）点为单位的半径（以公里计）。（双重，默认：
-1)
5.20. 推特流源
-
这
过滤 API返回与一个或多个过滤谓词匹配的公共状态。 多个参数支持使用单一连接流媒体API。 提示：该跟踪,跟随和地点场与运算符结合！ 查询track=foo和跟随=1234回复 推文匹配测试或者由用户创建1234. -
这
示例API返回所有公共状态的一小部分随机样本。 默认访问级别返回的推文是相同的，所以如果两个不同的客户端连接到该端点，它们会看到相同的推文。
默认访问层允许最多400条关键词、5000个关注用户ID和25个0.1-360度位置框。
5.20.1. 期权
按前缀分组的属性：
Twitter.connection
- 访问Tokens
-
你的TwitterTokens。（字符串，默认：
<没有>) - 访问Tokens秘密
-
你的推特Tokens秘密。（字符串，默认：
<没有>) - 消费者密钥
-
你的推特密钥。（字符串，默认：
<没有>) - 消费者秘密
-
你的推特秘密。（字符串，默认：
<没有>) - 调试支持
-
启用 Twitter4J 调试模式。（布尔值，默认：
false) - raw-json
-
启用缓存由 Twitter API 返回的原始（原始）JSON 对象。当设置为 False 时，结果将使用 Twitter4J 的 json 表示。设置为 True 时，结果将使用原始的 Twitter APIS json 表示。（布尔值，默认：
true)
推特.stream.filter
- 计数
-
表示在切换到直播前需要直播的之前状态数量。（整数，默认：
0) - Filter级
-
过滤器级别限制了流中出现的推文，仅限于具有最小 filterLevel 属性值的推文。要么是无，要么是低，要么是中等。（FilterLevel，默认：
<没有>) - 跟随
-
通过ID指定用户接收公共推文。（列表<长>，默认：
<没有>) - 语言
-
指定流的推文语言。（List<String>，默认：
<没有>) - 地点
-
需要追踪的位置。内部表示为二维数组。边界框无效：52.38、4.90、51.51、-0.12。第一对必须是盒子的西南角（List<BoundingBox>，默认：
<没有>) - 跟踪
-
指定需要追踪的关键词。（List<String>，默认：
<没有>)
推特.stream
- 类型
-
<缺少文档>（StreamType，默认：
<没有>，可能的值：样本,Filter,消防水带,链接)
5.21. Websocket 源代码
这Websocket通过Web Socket生成消息的源代码。
5.21.1. 选项
按前缀分组的属性：
websocket.supplier
- 允许的起源
-
允许的起源。（字符串，默认：
*) - 路径
-
服务器WebSocket处理器被暴露的路径。（字符串，默认：
/websocket)
websocket.supplier.sock-js
- 使
-
在服务器上启用SockJS服务。默认是“假”（布尔值，默认：
false)
5.21.2. 示例
为了验证websocket-source是否接收Websocket客户端的消息，可以使用以下简单的端到端设置。
第一步：开始写卡夫卡
第二步：部署websocket-source在某个具体端口，比如8080端口
步骤3：连接一个位于8080端口路径“/websocket”的Websocket客户端，发送一些消息。
你可以启动一个Kafka主机用户，看到那里的消息。
5.22. XMPP 来源
“xmpp”源使得从XMPP服务器接收消息成为可能。
5.22.1. 输入
无
5.22.2. 输出
有效载荷
-
字节[]
5.22.3. 选项
xmpp 源代码有以下选项：
按前缀分组的属性：
XMPP.工厂
- 主机
-
连接XMPP主机服务器。（字符串，默认：
<没有>) - 密码
-
连接用户的密码。（字符串，默认：
<没有>) - 端口
-
连接主机的端口。- 默认客户端端口：5222（整数，默认：
5222) - 资源
-
XMPP主机上的资源绑定。- 可以是空的，如果未设置服务器将生成一个（字符串，默认：
<没有>) - 安全模式
-
<缺少文档>（安全模式，默认：
<没有>，可能的值：必填,如果可能的话,禁用) - 服务名称
-
为XMPP域设置服务名称。（字符串，默认：
<没有>) - 订阅模式
-
<缺少文档>（订阅模式，默认：
<没有>，可能的值：accept_all,reject_all,手动) - 用户
-
连接用户应该连接。（字符串，默认：
<没有>)
XMPP.提供商
- 有效载荷表达式
-
<缺少文档>（表达式，默认：
<没有>) - 诗节Filter
-
<缺少文档>（StanzaFilter，默认：
<没有>)
另请参阅 Spring Boot 文档，了解代理连接和监听者属性的附加属性。
5.22.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps你可以在这里找到相应的活页夹项目。 然后你可以把光盘放进某个文件夹里，然后构建它：
$ ./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. 输出
有效载荷
-
字节[]
5.23.3. 选项
zeromq 源具有以下选项：
- zeromq.supplier.bind-port
-
绑定端口用于创建 ZeroMQ 套接字;0 选择一个随机端口。（整数，默认：
0) - zeromq.supplier.connect-url
-
连接到 ZeroMQ 套接字的连接 URL。（字符串，默认：
<没有>) - zeromq.supplier.consume-delay
-
当没有收到数据时，ZeroMQ 套接字的消耗延迟。（持续时间，默认：
1) - zeromq.supplier.socket-type
-
连接应该形成的插槽类型。（SocketType，默认：
<没有>，可能的值：双,酒馆,子,要求,代表,贩子,路由器,拉,推,XPUB,XSUB,流,客户端,服务器,收音机,菜,渠道,同辈,生,散射,收集) - zeromq.supplier.topics
-
订阅主题。（字符串[]，默认：
[])
另请参阅 Spring Boot 文档，了解代理连接和监听者属性的附加属性。
5.23.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps你可以在这里找到相应的活页夹项目。 然后你可以把光盘放进某个文件夹里，然后构建它：
$ ./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. 有效载荷
如果输入有效载荷是字节[]content-type 头是 JSON，则JsonBytesToMap函数尝试将该有效载荷反序列化为地图以便更好地表示聚合器函数输出的数据。
还有，这样的地图数据表示使得从下面提到的 SpEL 表达式中访问有效载荷内容变得容易。
否则（包括反序列化错误），输入有效载荷保持原样——目标应用配置将其转换为所需形式。
6.1.2. 选项
按前缀分组的属性：
聚合
- 集合体
-
聚合策略的SpEL表达式。默认是收集有效载荷。（表达式，默认：
<没有>) - 相关
-
相关性键的SpEL表达式。默认使用 correlationId 头部。（表达式，默认：
<没有>) - 小组暂停
-
SpEL表达式用于超时至到期的未完成组。（表达式，默认：
<没有>) - 消息存储实体
-
持久性消息存储实体：RDBMS 中的表前缀，MongoDb 中的集合名称等。（字符串，默认：
<没有>) - 消息存储类型
-
消息存储类型。（字符串，默认：
<没有>) - 释放
-
发布策略的SpEL表达式。默认设置基于序列大小头部。（表达式，默认：
<没有>)
spring.data.mongodb
- 额外主持人
-
额外的服务器主机。如果没有指定“host”，则无法用URI设置。其他主机将使用默认的Mongo端口27017。如果你想用不同的端口，可以用“host：port”语法。（List<String>，默认：
<没有>) - 认证数据库
-
认证数据库名称。（字符串，默认：
<没有>) - 自动索引创建
-
是否启用自动索引创建。（布尔值，默认：
<没有>) - 数据库
-
数据库名称。覆盖URI数据库。（字符串，默认：
<没有>) - 场命名策略
-
字段命名策略的完全限定名称。（职业<？>，默认：
<没有>) - 主机
-
Mongo服务器主机。无法用URI设置。（字符串，默认：
<没有>) - 密码
-
Mongo服务器的登录密码。无法用URI设置。（字符[]，默认：
<没有>) - 端口
-
Mongo服务器端口。无法用URI设置。（整数，默认：
<没有>) - 复制集名称
-
集群的副本集名称要求。无法用URI设置。（字符串，默认：
<没有>) - 乌里
-
Mongo数据库URI。覆盖主机、端口、用户名和密码。（字符串，默认：
mongodb://localhost/test) - 用户名
-
登录Mongo服务器的用户。无法用URI设置。（字符串，默认：
<没有>) - UUID 表示
-
将 UUID 转换为 BSON 二进制值时使用的表示方式。（UuidRepresentation，默认：
Java-遗产，可能的值：未具体说明,标准,C_SHARP_LEGACY,JAVA_LEGACY,PYTHON_LEGACY)
spring.data.mongodb.gridfs
- 桶
-
GridFS 桶名称。（字符串，默认：
<没有>) - 数据库
-
GridFS 数据库名称。（字符串，默认：
<没有>)
spring.data.mongodb.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<没有>) - 启用
-
是否启用SSL支持。除非另有说明，否则若提供了“捆绑”，则自动启用。（布尔值，默认：
<没有>)
spring.data.redis
- 客户名
-
客户端名称应与客户端 SETNAME 连接时设置。（字符串，默认：
<没有>) - 客户端类型
-
客户端类型。默认情况下，根据类路径自动检测。（ClientType，默认：
<没有>，可能的值：生菜,杰迪斯) - 连接超时
-
连接超时。（持续时间，默认：
<没有>) - 数据库
-
连接工厂使用的数据库索引。（整数，默认：
0) - 主机
-
Redis服务器主机。（字符串，默认：
本地主持) - 密码
-
Redis 服务器的登录密码。（字符串，默认：
<没有>) - 端口
-
Redis服务器端口。（整数，默认：
6379) - 超时
-
读“暂停”。（持续时间，默认：
<没有>) - 网址
-
连接URL。覆盖主机、端口、用户名和密码。示例：redis://user：[email protected]：6379（字符串，默认：
<没有>) - 用户名
-
登录Redis服务器的用户名。（字符串，默认：
<没有>)
spring.data.redis.cluster
- 最大重定向
-
执行集群命令时应执行的最大重定向次数。（整数，默认：
<没有>) - 节点
-
用逗号分隔的“host：port”对列表，用于引导。这代表一个“初始”集群节点列表，且至少需要一个条目。（List<String>，默认：
<没有>)
spring.data.redis.jedis.pool
- 启用
-
是否启用该池。如果有“commons-pool2”可用，则自动启用。在Jedis中，在哨兵模式下池化是隐式启用的，且该设置仅适用于单节点设置。（布尔值，默认：
<没有>) - 最大主动
-
池在特定时间内可分配的最大连接数。为无限制使用负值。（整数，默认：
8) - 最大怠速
-
池中“空闲”连接的最大数量。使用负值表示空闲连接数量无限。（整数，默认：
8) - 等一下
-
连接分配应在池耗尽后才触发异常的最长阻塞时间。使用负值无限期阻塞。（持续时间，默认：
-1毫秒) - 最小怠速
-
目标是池中最少需要维持的空闲连接数。该设置仅在该设置和驱逐运行间距均为正值时生效。（整数，默认：
0) - 驱逐间隔时间
-
空闲对象evictor线程运行间隔时间。当正时，空闲对象Evictor线程开始，否则不会执行空闲对象驱逐。（持续时间，默认：
<没有>)
spring.data.redis.lettuce.pool
- 启用
-
是否启用该池。如果有“commons-pool2”可用，则自动启用。在Jedis中，在哨兵模式下池化是隐式启用的，且该设置仅适用于单节点设置。（布尔值，默认：
<没有>) - 最大主动
-
池在特定时间内可分配的最大连接数。为无限制使用负值。（整数，默认：
8) - 最大怠速
-
池中“空闲”连接的最大数量。使用负值表示空闲连接数量无限。（整数，默认：
8) - 等一下
-
连接分配应在池耗尽后才触发异常的最长阻塞时间。使用负值无限期阻塞。（持续时间，默认：
-1毫秒) - 最小怠速
-
目标是池中最少需要维持的空闲连接数。该设置仅在该设置和驱逐运行间距均为正值时生效。（整数，默认：
0) - 驱逐间隔时间
-
空闲对象evictor线程运行间隔时间。当正时，空闲对象Evictor线程开始，否则不会执行空闲对象驱逐。（持续时间，默认：
<没有>)
spring.data.redis.lettuce
- 关机超时
-
关机暂停。（持续时间，默认：
100毫秒)
spring.data.redis.sentinel
- 主人
-
Redis 服务器名称。（字符串，默认：
<没有>) - 节点
-
“host：port”对的逗号分隔列表。（List<String>，默认：
<没有>) - 密码
-
用哨兵认证的密码。（字符串，默认：
<没有>) - 用户名
-
登录用户名用于用Sentinel认证。（字符串，默认：
<没有>)
spring.data.redis.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<没有>) - 启用
-
是否启用SSL支持。除非另有说明，否则若提供了“捆绑”，则自动启用。（布尔值，默认：
<没有>)
spring.datasource
- 驾驶员类别名称
-
JDBC驱动的名称是完全限定的。默认情况下，基于URL自动检测。（字符串，默认：
<没有>) - 密码
-
数据库的登录密码。（字符串，默认：
<没有>) - 网址
-
数据库的JDBC网址。（字符串，默认：
<没有>) - 用户名
-
数据库用户名登录。（字符串，默认：
<没有>)
6.3. 滤波处理器
过滤处理器使应用程序能够检查进入的有效载荷，然后对其应用一个谓词，决定记录是否需要继续。
例如，如果输入的有效载荷类型为字符串如果你想过滤掉少于五个字符的内容，可以按照下面的方式运行过滤处理器。
Java -jar filter-processor-kafka-<version>.jar --filter.function.expression=payload.length（） > 4
如果你想在RabbitMQ上运行Kafka，可以把它改成Rabbit。
6.3.1. 有效载荷
你可以将任意类型传递为有效载荷，然后对其应用 SpEL 表达式进行过滤。
如果输入类型是字节[]内容类型设置为文本/纯文字或application/json然后应用程序将字节[]到字符串.
6.3.2. 选项
- 滤波功能
-
布尔 SpEL 表达式用于应用请求消息以进行过滤。（表达式，默认：
<没有>)
6.4. Groovy Processor
一个对消息应用Groovy脚本的处理器。
6.4.1. 选项
groovy-processor处理器具备以下选项：
- groovy-processor.script
-
引用用于处理消息的脚本。（资源，默认：
<没有>) - groovy-processor.variables
-
变量绑定作为新行分隔串的名称-值对，例如 'foo=bar\n baz=car'。（属性，默认：
<没有>) - groovy-processor.variables-location
-
包含自定义脚本变量绑定的属性文件的位置。（资源，默认：
<没有>)
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。（属性，默认：
<没有>) - header.enricher.overwrite
-
设置为 true，以覆盖所有现有的消息头。（布尔值，默认：
false)
6.6. Http请求处理器
一个处理器应用，向HTTP资源发出请求，并将响应体作为消息载荷发出。
6.6.1. 输入
头
任何必修的HTTP头必须通过头或头部表达式财产。请参见下方示例。
头部值也可以用于构造：
-
请求主体在
身体表情财产。 -
HTTP方法在
http-方法表达式财产。 -
在
URL 表达式财产。
有效载荷
有效载荷默认作为POST请求的请求体，可以是任何Java类型。 对于 GET 请求，应该是空字符串。 有效载荷还可用于构建：
-
请求主体在
身体表情财产。 -
HTTP方法在
http-方法表达式财产。 -
在
URL 表达式财产。
底层Web客户端支持Jackson JSON序列化，必要时支持任何请求和响应类型。
这期望反应型属性String.class默认情况下，可以设置为你应用类路径中的任何类。
注意，用户自定义的负载类型需要向你的 pom 文件添加必要的依赖。
6.6.2. 输出
头
没有任何 HTTP 消息头映射到出站消息。
有效载荷
原始输出对象是 ResponseEntity<？>其任意字段（例如，身体,头）或访问器方法（状态代码）可以作为回复表达式.
默认情况下，出站消息有效载荷是响应体。
注意 ResponseEntity（由表达式#root）默认情况下不能被Jackson反序列化，但可以被表示为哈希图.
6.6.3. 期权
http-request 处理器具备以下选项：
6.6.4. 期权
按前缀分组的属性：
http.request
- 身体表情
-
一个SpEL表达式，用于从收到的消息中推导出请求体。（表达式，默认：
<没有>) - 期望反应型
-
用来解读回应的类型。（职业<？>，默认：
<没有>) - 头部表达式
-
用于推导 HTTP 头部映射的 SpEL 表达式。（表达式，默认：
<没有>) - http-方法表达式
-
一个SpEL表达式，用于从收到的消息中推导出请求方法。（表达式，默认：
<没有>) - 回复表达式
-
用于计算最终结果的 SpEL 表达式，应用于整个 http {@link org.springframework.http.ResponseEntity}。（表达式，默认：
<没有>) - 超时
-
请求超时，毫秒级。（长，默认：
30000) - URL 表达式
-
对入件消息的 SpEL 表达式来确定应使用的 URL。（表达式，默认：
<没有>)
spring.codec
- 日志-请求-详情
-
是否在DEBUG级别记录表单数据，在TRACE级别记录头部。（布尔值，默认：
false) - 最大内存容量
-
限制每当需要聚合输入流时可缓冲的字节数。这仅适用于自动配置的WebFlux服务器和Web客户端实例。默认情况下，这个设置不被设置，因此适用各个编解码器的默认值。大多数编解码器默认限制为256K。（DataSize，默认：
<没有>)
6.7. 图像识别处理器
使用初始化模型进行分类的处理器 实时图像被分割成不同类别（例如标签）。
模型的输入是图像，作为二进制数组。
输出为以下格式的 JSON 消息：
{
"labels" : [
{"giant panda":0.98649305}
]
}结果包含识别类别的名称（例如标签）以及图像代表该类别的置信度（例如置信度）。
如果响应-扣押当值设置为大于1时，结果将包括顶部响应-扣押可能的标签。例如响应大小=3将回归：
{
"labels": [
{"giant panda":0.98649305},
{"badger":0.010562794},
{"ice bear":0.001130851}
]
}6.7.1. 有效载荷
如果输入类型是字节[]内容类型设置为应用/八位元组流，然后应用程序处理输入字节[]图像输入并输出增强字节[]图像负载和json头部。
6.7.2. 期权
- image.recognition.cache-model
-
缓存预训练的张量流模型。（布尔值，默认：
true) - image.recognition.debug-output
-
<缺少文档>（布尔值，默认：
false) - image.recognition.debug-output-path
-
<缺少文档>（字符串，默认：
image-recognition-result.png) - image.recognition.model
-
预训练张量流图像识别模型。请注意，模型必须与所选模型类型相匹配！（字符串，默认：
https://storage.googleapis.com/mobilenet_v2/checkpoints/mobilenet_v2_1.4_224.tgz#mobilenet_v2_1.4_224_frozen.pb) - image.recognition.model-type
-
支持三种不同的预训练张量流图像识别模型：Inception、MobileNetV1 和 MobileNetV2 1。Inception Graph 使用“input”作为输入，“output”作为输出。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 - 图以“input”为输入，输出为“MobilenetV1/Predictions/Reshape_1”。（ModelType，默认：
<没有>，可能的值：初始,移动网1,移动网2) - image.recognition.normalized-image-size
-
图像尺寸归一化。（整数，默认：
224) - 图像。识别。响应大小
-
识别图像数量。（整数，默认：
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 -
对象.检测.标签:storage.googleapis.com/scdf-tensorflow-models/object-detection/mscoco_label_map.pbtxt -
object.detection.with-masks:false
下图展示了一个Spring云数据流流式流水线，实时预测输入图像流中的对象类型。
处理器的输入是一个图像字节数组，输出是一个增强后的图像和一个称为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. 有效载荷
入门类型为字节[]，内容类型为应用/八位元组流.处理器处理输入字节[]图像并输出增强版字节[]图像负载和JSON头（detected_objects).
6.8.2. 期权
- object.detection.cache-model
-
<缺少文档>（布尔值，默认：
true) - 对象。检测。信心
-
<缺少文档>（浮点，默认：
0.4) - object.detection.debug-output
-
<缺少文档>（布尔值，默认：
false) - object.detection.debug-output-path
-
<缺少文档>（字符串，默认：
object-detection-result.png) - 对象.检测.标签
-
标签为上呼吸道感染。（字符串，默认：
https://storage.googleapis.com/scdf-tensorflow-models/object-detection/mscoco_label_map.pbtxt) - object.detection.model
-
预训练张量流对象检测模型。（字符串，默认：
https://download.tensorflow.org/models/object_detection/ssdlite_mobilenet_v2_coco_2018_05_09.tar.gz#frozen_inference_graph.pb) - object.detection.response-size
-
<缺少文档>（整数，默认：
<没有>) - object.detection.with-masks
-
<缺少文档>（布尔值，默认：
false)
6.9. 语义分割处理器
基于最先进的DeepLab张量流模型的图像语义分割。
这语义分割是将图像中的每个像素与类别标签（如花、人、道路、天空、海洋或汽车）关联起来的过程。
与实例细分，产生实例感知区域掩码，即语义分割产生类感知的掩码。
用于实现实例细分建议咨询对象检测服务。
6.9.1. 有效载荷
入门类型为字节[]，内容类型为应用/八位元组流.处理器处理输入字节[]图像和输出增强字节[]图像负载和json头部。
处理器的输入是一个图像字节数组，输出是一个增强图像字节数组和一个 JSON 头部semantic_segmentation采用此格式：
[
[ 0, 0, 0 ],
[ 127, 127, 127 ],
[ 255, 255, 255 ],
...
]输出头json格式表示从输入图像计算出的彩色像素映射。
6.9.2. 期权
- 语义.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
-
计算出的分割掩码图像的α色。（浮点，默认：
0.45) - semantic.segmentation.model
-
预训练的张量流语义分割模型。（字符串，默认：
https://download.tensorflow.org/models/deeplabv3_mnv2_cityscapes_train_2018_02_05.tar.gz#frozen_inference_graph.pb) - semantic.segmentation.output-type
-
指定输出图像类型。你可以返回带有计算出的遮罩覆盖的输入图像，或者只返回遮罩。（OutputType，默认：
<没有>，可能的值：混合,面具)
6.10. 剧本处理器
处理器用脚本转换消息。剧本正体直接提供 作为房产价值。脚本的语言可以被指定（groovy/javascript/ruby/python）。
6.10.1. 选项
脚本处理器处理器具备以下选项：
- 脚本处理者.language
-
文字语言在脚本属性中。支持：groovy、javascript、ruby、python。（字符串，默认：
<没有>) - 脚本处理器.脚本
-
剧本文本。（字符串，默认：
<没有>) - 脚本处理器变量
-
变量绑定作为新行分隔串的名称-值对，例如 'foo=bar\n baz=car'。（属性，默认：
<没有>) - script-processor.variables-location
-
包含自定义脚本变量绑定的属性文件的位置。（资源，默认：
<没有>)
6.11. 分流处理器
分流器应用基于 Spring Integration 中同名的概念，允许将单条消息拆分成多个独立消息。
处理器使用一个函数，对留言<？>作为输入，然后产生列表<消息<？>作为基于各种性质的输出（见下文）。
你可以用SpEL表达式或分隔符来指定如何拆分收到的消息。
6.11.1. 有效载荷
-
来袭有效载荷——
留言<？>
如果输入类型是字节[]内容类型设置为文本/纯文字或application/json然后应用程序将字节[]到字符串.
-
即将发射的有效载荷——
列表<消息<？>
6.11.2. 选项
- splitter.apply-sequence
-
在头部添加相关/序列信息，便于后续聚合。（布尔值，默认：
true) - splitter.charset
-
在将基于文本的文件中字节转换为字符串时使用的字元集。（字符串，默认：
<没有>) - splitter.delimiters
-
当表达式为空时，分隔符用于分区{@link字符串}有效载荷分符。（字符串，默认：
<没有>) - 分流者。表达
-
用于拆分有效载荷的SpEL表达式。（字符串，默认：
<没有>) - splitter.file-markers
-
设置为true或false，以便使用{@code FileSplitter}（按行分割文本文件），该文件包含（或不包含）文件的开头/结尾标记。（布尔值，默认：
<没有>) - 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
-
一个需要应用的特殊语言表达。（字符串，默认：
<没有>)
6.13. Twitter 趋势和趋势位置处理器
能够返回热门话题或热门话题的位置的处理器。
这推特。趋势。趋势-查询类型属性允许选择查询类型。
6.13.2. 检索趋势位置
对于该模式集合推特。趋势。趋势-查询类型自趋势位置.
按地点获取完整或附近的热门话题列表。
如果纬度,经度只要处理器执行趋势可用API，并返回Twitter拥有热门话题信息的位置，则不包含参数。
如果纬度,经度参数前提是处理器执行趋势最近 API，并返回 Twitter 拥有趋势话题信息且最接近指定位置的位置。
响应是一个数组地点该信息编码了该地点的 WOEID 以及一些人类可读的信息，比如该位置所属的官方名称和国家。
6.13.3. 期权
按前缀分组的属性：
推特。趋势。最近
- 纬度
-
如果提供长参数，可用的趋势位置将按距离坐标对的距离排序。经度的有效范围为-180.0至+180.0（西为负，东为正），包括在内。（表达式，默认：
<没有>) - 离子
-
如果提供纬度参数，可用的趋势位置将按距离坐标对的距离排序。经度的有效范围为-180.0至+180.0（西为负，东为正），包括在内。（表达式，默认：
<没有>)
推特趋势
- 位置标识
-
雅虎！地球上的哪里，地址ID以返回趋势信息。通过使用 1 作为 WOEID 可获得全球信息。（表达式，默认：
有效载荷) - 趋势查询类型
-
<缺少文档>（TrendQueryType，默认：
<没有>，可能的值：趋势,趋势位置)
7. 沉没
7.1. 卡桑德拉·辛克
该汇入应用会将收到的每条消息的内容写入Cassandra。
它期望收到 JSON String 的负载，并用其属性映射到表列。
7.1.1. 有效载荷
一个 JSON 字符串或字节数组，表示要持久化的实体（或实体列表）。
7.1.2. 期权
Cassandra水槽有以下选项：
按前缀分组的属性：
cassandra.cluster
- 创建密钥空间
-
在应用启动时标记键空间（或不创建）。（布尔值，默认：
false) - 实体基包
-
用于扫描带有表注释的实体的基础包。（字符串[]，默认：
[]) - 初始化脚本
-
资源中用CQL脚本（用';'分隔）来初始化密钥空间模式。（资源，默认：
<没有>) - skip-ssl-validation
-
标记以验证服务器的SSL证书。（布尔值，默认：
false)
Cassandra
- 一致性层级
-
写作的一致性级别。（ConsistencyLevel，默认：
<没有>) - ingest-query
-
获取Cassandra查询。（字符串，默认：
<没有>) - 查询类型
-
Cassandra Sink 的 QueryType。（类型，默认：
<没有>，可能的值：插入,更新,删除,陈述) - 语句表达式
-
Cassandra 查询中的表达式 DSL 风格。（表达式，默认：
<没有>) - TTL
-
WriteOptions 的“存活时间”选项。（整数，默认：
0)
Spring。卡桑德拉
- 压缩
-
压缩由Cassandra二进制协议支持。（压缩，默认：
没有，可能的值：LZ4,活泼,没有) - 配置
-
配置文件的位置。（资源，默认：
<没有>) - 接触点
-
集群节点地址形式为“host：port”，或简单地使用配置好的端口“host”。（List<String>，默认：
[127.0.0.1:9042]) - keyspace-name
-
使用密钥空间名称。（字符串，默认：
<没有>) - 本地数据中心
-
被视为“本地”的数据中心。联系点应来自该数据中心。（字符串，默认：
<没有>) - 密码
-
服务器登录密码。（字符串，默认：
<没有>) - 端口
-
如果某个接触点没有指定端口，可以使用端口。（整数，默认：
9042) - 模式动作
-
启动时需要采取的模式作。（字符串，默认：
没有) - 会话名称
-
卡桑德拉的会议名称。（字符串，默认：
<没有>) - 用户名
-
登录服务器用户。（字符串，默认：
<没有>)
Spring.cassandra.connection。
- 连接超时
-
在建立驱动连接时使用超时。（持续时间，默认：
5) - init-query-timeout
-
超时用于初始化过程中运行的内部查询，连接刚打开后。（持续时间，默认：
5)
春.cassandra.controlconnection
- 超时
-
用于控制查询的超时。（持续时间，默认：
5)
Spring。Cassandra.pool。
- 心跳间隔
-
心跳间隔，之后会通过空闲连接发送消息，以确保它仍然活着。（持续时间，默认：
30多岁) - 闲置超时
-
在移除空闲连接之前，空闲超时。（持续时间，默认：
5)
春季.cassandra.request
- 一致性
-
查询一致性等级。（DefaultConsistencyLevel，默认：
<没有>，可能的值：任何,一,二,三,法定人数,都,LOCAL_ONE,LOCAL_QUORUM,EACH_QUORUM,串行,LOCAL_SERIAL) - 页码大小
-
单次网络往返中会同时检索多少行。（整数，默认：
5000) - 串行一致性
-
查询串行一致性级别。（DefaultConsistencyLevel，默认：
<没有>，可能的值：任何,一,二,三,法定人数,都,LOCAL_ONE,LOCAL_QUORUM,EACH_QUORUM,串行,LOCAL_SERIAL) - 超时
-
Drivers等待请求完成的时间。（持续时间，默认：
2)
春.cassandra.request.throttler
- 排水间隔
-
节流器尝试取消队列请求的频率。设置足够高，使每次尝试处理队列中的多个条目，但不会延迟请求过久。（持续时间，默认：
<没有>) - max-concurrent-requests
-
允许并行执行的最大请求数。（整数，默认：
<没有>) - 最大队列大小
-
超过限速阈值时，可以排队的最大请求数。（整数，默认：
<没有>) - 每秒最大请求数
-
最高允许的请求速率。（整数，默认：
<没有>) - 类型
-
请求限速类型。（ThrottlerType，默认：
没有，可能的值：CONCURRENCY_LIMITING,RATE_LIMITING,没有)
春.cassandra.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<没有>) - 启用
-
是否启用SSL支持。（布尔值，默认：
<没有>)
7.2. 分析汇
Sink 应用构建在 Analytics Consumer 之上，能够从输入消息中计算分析数据，并将分析数据作为指标发布给各种监控系统。它利用微米库在最流行的监控系统中提供统一的编程体验，并揭示了Spring表达式语言（SpEL）特性，用于定义如何从输入数据计算度量名称、值和标签。
分析汇可以生成两种指标类型：
米（如计数器或规范）由其唯一标识名称和尺寸（尺寸和标签这两个术语可以互换使用。）维度允许对特定命名的指标进行切片，以深入分析和推理数据。
作为度量唯一由其标识的名称和尺寸你可以为每个指标分配多个标签（e.g. key/值对），但不能随意更改这些标签！像Prometheus这样的监控系统如果同名指标使用不同的标签组，就会抱怨。 |
使用该analytics.name或analytics.name-expression属性设置了输出分析指标的名称。如果没有设置，指标名称默认为应用程序名称。
使用该analytics.tag.expression.<TAG_NAME>=<TAG_VALUE>，属性，用于在你的指标中添加一个或多个标签。这TAG_NAME属性定义中使用的指标将作为标签名称出现。TAG_VALUE是SpEL表达式，能够动态计算来自接收消息的标签值。
这SpEL表达式使用头和有效载荷关键词用于访问消息的头部和有效载荷值。
你可以使用文字（例如：“固定价值”）以设置固定值的标签。 |
所有Stream Applications一出厂就支持三种最受欢迎的监控系统，波,普罗 米修斯和InfluxDB你可以声明式地启用它们。
你可以通过在分析汇应用。
请访问春季云数据流监控，获取详细的监控系统配置说明。以下简要内容可以帮助你开始。
-
要启用普罗米修斯电表注册表，请设置以下属性。
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-
要启用波前表注册表，请设置以下属性。
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}
如果启用了数据流服务器监控，那么分析汇将重复使用提供的指标配置。 |
下图展示了如何分析汇帮助收集企业内部信息，用于股票交易所和实时管道。
7.2.1. 有效载荷
收到的消息可以包含任何类型的有效载荷。
7.2.2. 选项
按前缀分组的属性：
分析学
- 数量表达
-
一个 SpEL 表达式用于计算输出度量值（例如数量）。默认是1.0（表达式，默认：
<没有>) - 米型
-
用于向后台报告指标的微米类型。（MeterType，默认：
<没有>，可能的值：计数器,轨距) - 名称
-
输出指标的名称。“name”和“nameExpression”是互斥的。只能设置其中一个。（字符串，默认：
<没有>) - 名称表达式
-
一个 SpEL 表达式，用于从输入消息中计算输出度量名称。“name”和“nameExpression”是互斥的。只能设置其中一个。（表达式，默认：
<没有>)
analytics.tag
- 表达
-
通过SpEL表达式计算标签。单个 SpEL 表达式可以产生数组值，这又意味着不同的名称/值标签。每个名字/值标签都会生成一个独立的计量器增量。标签表达格式为：analytics.tag.expression。[标签名]=[SpEL表达式]（Map <String， Expression>，默认：
<没有>) - 固定
-
弃用：请使用带有字面SpEL表达式的analytics.tag.expression。自定义，固定标签。这些标签有固定值，创建一次后会随发布的指标一起发送。定义固定标签的惯例是：<code>analytics.tag.fixed。[标签名]=[标签值] </代码>（Map<String， String>，默认：
<没有>)
7.3. 弹性搜索汇
一个能将文档索引到Elasticsearch的Sink。
这个 Elasticsearch sink 只支持索引 JSON 文档。
它从输入目的地获取数据，然后将其索引到 Elasticsearch。
输入数据可以是普通的 JSON 字符串，也可以是java.util.Map代表 JSON。
它也接受Elasticsearch提供的数据XContentBuilder.
不过这种情况很少见，因为中间件不太可能像XContentBuilder.
这主要用于直接调用消费者。
7.3.1. 期权
Elasticsearch 汇具备以下选项：
按前缀分组的属性：
elasticsearch.consumer
- 异步
-
表示索引作是否异步。默认情况下，索引是同步完成的。（布尔值，默认：
false) - 批次大小
-
每个请求需要索引的项目数量。默认是1。对于大于 1 的值，将使用 批量索引 API。（整数，默认：
1) - 小组暂停
-
超时以毫秒计，之后在批量索引激活时会冲洗消息组。默认为 -1，意味着不会自动刷新空闲消息组。（长，默认：
-1) - 身份证
-
文档的索引ID。如果设置为，INDEX_ID头值会在每条消息上覆盖该属性。（表达式，默认：
<没有>) - 指数
-
索引名称。如果设置为，INDEX_NAME头值会在每条消息上覆盖该属性。（字符串，默认：
<没有>) - 路由
-
指示要路由的分片。如果未提供，Elasticsearch 将默认使用文档 ID 的哈希值。（字符串，默认：
<没有>) - 暂停秒
-
分片可用的时间暂停。如果没有设置，默认情况下由 Elasticsearch 客户端设置 1 分钟。（长，默认：
0)
spring.elasticsearch
- 连接超时
-
连接超时用于与 Elasticsearch 通信。（持续时间，默认：
1) - 密码
-
用于Elasticsearch认证的密码。（字符串，默认：
<没有>) - 路径前缀
-
在发送给 Elasticsearch 的每个请求路径中添加前缀。（字符串，默认：
<没有>) - 插座-保持生命
-
是否启用客户端和Elasticsearch之间的套接字保持活力。（布尔值，默认：
false) - 套接字超时
-
与 Elasticsearch 通信时使用的套接字超时。（持续时间，默认：
30多岁) - 乌里斯
-
使用Elasticsearch实例的逗号分隔列表。（List<String>，默认：
[http://localhost:9200]) - 用户名
-
用户名用于Elasticsearch认证。（字符串，默认：
<没有>)
Spring.elasticsearch.restclient.sniffer
- 失效后延迟
-
在失败后安排的嗅探执行延迟。（持续时间，默认：
1米) - 间隔
-
连续普通嗅探执行之间的间隔。（持续时间，默认：
5米)
Spring.elasticsearch.restclient.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<没有>)
7.3.2. 运行该水槽的示例
-
摘自文件夹
弹性搜索汇:./mvnw 清洁封装 -
CD应用
-
cd 到正确的绑定器生成应用（Kafka 或 RabbitMQ）
-
./mvnw 清洁封装 -
确保你运行着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 -
字节[] -
字符串
7.4.2. 选项
这文件汇入有以下选项：
- file.consumer.binary
-
一个标志用于指示是否应在写入后添加换行。（布尔值，默认：
false) - file.consumer.charset
-
写文本内容时使用的字符集。（字符串，默认：
UTF-8) - file.consumer.directory
-
目标文件的父目录。（文件，默认：
<没有>) - file.consumer.directory-expression
-
计算目标文件父目录的表达式。（字符串，默认：
<没有>) - file.consumer.mode
-
如果目标文件已经存在，则使用 FileExistsMode 来使用。（FileExistsMode，默认：
<没有>，可能的值：附加,APPEND_NO_FLUSH,失败,忽视,取代,REPLACE_IF_MODIFIED) - file.consumer.name
-
目标文件的名称。（字符串，默认：
文件消费者) - file.consumer.name-expression
-
用于计算目标文件名称的表达式。（字符串，默认：
<没有>) - file.consumer.suffix
-
文件名后加上后缀。（字符串，默认：
<空字符串>)
7.5. FTP 汇
FTP sink 是一个简单的选项，可以将收到的消息推送文件到 FTP 服务器。
它使用了一个FTP-出站适配器因此，收到的消息可以是java.io.file对象，a字符串（文件内容）
或字节（文件内容也包括）
使用这个水槽需要用户名和密码才能登录。
默认情况下，Spring 集成会使用o.s.i.file.DefaultFileNameGenerator如果没有指定。DefaultFileNameGenerator将决定文件名
基于file_name在消息头，或者如果 的有效载荷消息已经是java.io.file那么它就会
使用该文件的原始名称。 |
7.5.1. 头部
-
file_name（见上文注释）
7.5.2. 有效载荷
-
java.io.file -
java.io.InputStream -
字节[] -
字符串
7.5.3. 输出
不行（写入FTP服务器）。
7.5.4. 期权
FTP Sink 有以下选项：
按前缀分组的属性：
ftp.consumer
- 自动-创建-导演
-
是否创建远程目录。（布尔值，默认：
true) - 文件名表达式
-
一个用于生成远程文件名的 SpEL 表达式。（字符串，默认：
<没有>) - 模式
-
如果远程文件已经存在，则需要采取相应的作。（FileExistsMode，默认：
<没有>，可能的值：附加,APPEND_NO_FLUSH,失败,忽视,取代,REPLACE_IF_MODIFIED) - 远程指令
-
远程FTP目录。（字符串，默认：
/) - 远程文件分隔器
-
远程文件分离器。（字符串，默认：
/) - 临时-远程-DIR
-
一个临时目录，如果“#isUseTemporaryFilename（）”为真，文件将被写入。（字符串，默认：
/) - tmp-file-后缀
-
转移过程中使用的后缀。（字符串，默认：
.tmp) - 使用临时文件名
-
是否写入临时文件并重命名。（布尔值，默认：
true)
FTP.factory
- 缓存会话
-
缓存会话。（布尔值，默认：
<没有>) - 客户端模式
-
用于FTP会话的客户端模式。（客户端模式，默认：
<没有>，可能的值：积极,被动) - 主机
-
服务器的主机名。（字符串，默认：
本地主持) - 密码
-
连接服务器的密码。（字符串，默认：
<没有>) - 端口
-
服务器的移植。（整数，默认：
21) - 用户名
-
连接服务器的用户名。（字符串，默认：
<没有>)
7.6. JDBC汇入
JDBC sink 允许你将进入的有效载荷持久化到关系数据库数据库中。
这jdbc.consumer.columns性质表示COLUMN_NAME[：EXPRESSION_FOR_VALUE]哪里EXPRESSION_FOR_VALUE（连同冒号）是可选的。
在这种情况下，该值通过生成的表达式进行计算，如有效载荷。COLUMN_NAME这样我们就能直接映射对象属性到表列。
例如，我们有一个像这样的JSON有效载荷：
{
"name": "My Name",
"address": {
"city": "Big City",
"street": "Narrow Alley"
}
}所以，我们可以将其代入到表中，且名称,城市和街结构基于以下构型：
--jdbc.consumer.columns=name,city:address.city,street:address.street该汇入支持批处理插入，前提是底层JDBC驱动支持。
批量插入通过批次大小和闲置超时性能：
收到的消息会被聚合到批次大小消息存在后，作为批次插入。
如果闲置超时毫秒过去后没有新消息，即使聚合批次小于批次大小，限制最大延迟。
该模块还使用 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) - 表名
-
写入的表格名称。（字符串，默认：
消息)
spring.datasource
- 驾驶员类别名称
-
JDBC驱动的名称是完全限定的。默认情况下，基于URL自动检测。（字符串，默认：
<没有>) - 密码
-
数据库的登录密码。（字符串，默认：
<没有>) - 网址
-
数据库的JDBC网址。（字符串，默认：
<没有>) - 用户名
-
数据库用户名登录。（字符串，默认：
<没有>)
7.7. 阿帕奇·卡夫卡沉没
该模块向Apache Kafka发布消息。
7.7.1. 期权
卡夫卡水槽有以下选项：
（请参阅 Spring Boot 文档中关于 Spring for Apache Kafka 配置属性的相关内容）
按前缀分组的属性：
卡夫卡出版社
- 钥匙
-
Kafka 记录密钥——如果提供了，则被 keyExpression 覆盖。（字符串，默认：
<没有>) - key-expression
-
一个将计算值为卡夫卡记录键的特殊字母表达式。（表达式，默认：
<没有>) - 映射头部
-
这些头部将被映射。（字符串[]，默认：
[*]) - 分区
-
Kafka 主题分区——如果提供，则被 partitionExpression 覆盖。（整数，默认：
<没有>) - 分区表达式
-
一个 SpEL 表达式，可以评估到 Kafka 主题划分。（表达式，默认：
<没有>) - 发送超时
-
Kafka制作人处理员应该等待多久才能收到发送作结果。默认为10秒。（持续时间，默认：
10分) - 同步
-
如果Kafka制作处理器应该以同步模式作，这点是正确的。（布尔值，默认：
false) - 时间戳
-
Kafka记录时间戳——如果提供了，则被timestampExpression覆盖。（长，默认：
<没有>) - 时间戳表达式
-
一个计算卡夫卡记录时间戳的特殊语言表达式。（表达式，默认：
<没有>) - 主题
-
卡夫卡主题——如果提供了，则被topicExpression覆盖。默认为KafkaTemplate.getDefaultTopic（）（字符串，默认：
<没有>) - 主题表达
-
一个针对卡夫卡主题进行评估的特殊英语表达式。（表达式，默认：
<没有>) - 使用模板转换器
-
是否使用模板的消息转换器来创建Kafka记录。（布尔值，默认：
false)
Spring。卡夫卡
- 自举服务器
-
用于建立与Kafka集群初始连接的host：port对列表。除非被覆盖，否则适用于所有组件。（List<String>，默认：
<没有>) - 客户端ID
-
发送请求时传递给服务器的ID。用于服务器端日志记录。（字符串，默认：
<没有>) - 性能
-
用于配置客户端的附加属性，这些属性是生产者和消费者共同的。（Map<String， String>，默认：
<没有>)
春季。卡夫卡。制片人
- 补丁
-
制作人要求领导者在确认请求完成前必须收到的确认数量。（字符串，默认：
<没有>) - 批次大小
-
默认批量大小。小批量会使批量处理变得不那么常见，并可能降低通量（批次大小为零则完全禁用批量处理）。（DataSize，默认：
<没有>) - 自举服务器
-
用于建立与Kafka集群初始连接的host：port对列表。对生产者来说，这覆盖了全球属性。（List<String>，默认：
<没有>) - 缓冲存储器
-
生产者可以用来缓冲等待发送到服务器的记录的总内存容量。（DataSize，默认：
<没有>) - 客户端ID
-
发送请求时传递给服务器的ID。用于服务器端日志记录。（字符串，默认：
<没有>) - 压缩型
-
为生产者生成的所有数据采用压缩类型。（字符串，默认：
<没有>) - 密钥串行器
-
键的串行器类。（职业<？>，默认：
<没有>) - 性能
-
用于配置客户端的额外生产者特有属性。（Map<String， String>，默认：
<没有>) - 重试
-
当大于零时，允许重试发送失败。（整数，默认：
<没有>) - 事务ID-前缀
-
当非空时，支持生产者的交易。（字符串，默认：
<没有>) - 值序列化器
-
串行器类用于值。（职业<？>，默认：
<没有>)
spring.kafka.template
- 默认主题
-
发送消息的默认主题。（字符串，默认：
<没有>) - 事务ID-前缀
-
交易ID前缀，覆盖生产工厂内的交易ID前缀。（字符串，默认：
<没有>)
7.8. 木头水槽
这日志Sink 使用应用日志器输出数据进行检查。
请理解这一点日志Sink 使用无类型处理程序，这会影响实际日志的执行方式。
这意味着如果内容类型是文本，那么原始有效载荷字节会被转换为字符串，否则原始字节会被记录。
更多信息请参见用户指南。
7.8.1. 期权
原木水槽有以下选项：
- log.expression
-
一个针对收到消息的 SpEL 表达式来评估为已记录的消息。（字符串，默认：
有效载荷) - log.level
-
记录消息的层级。（等级，默认：
<没有>，可能的值：致命,错误,警告,信息,调试,跟踪) - log.name
-
使用Logging机的名字。（字符串，默认：
<没有>)
7.9. MongoDB 汇入
该汇入应用将输入数据导入MongoDB。
该应用完全基于MongoDataAutoConfiguration，更多信息请参阅 Spring Boot MongoDB 支持。
7.9.1. 输入
有效载荷
-
任何POJO
-
字符串 -
字节[]
7.9.2. 期权
Mongodb水槽有以下选项：
按前缀分组的属性：
Mongodb.consumer
- 收集
-
MongoDB集合用于存储数据。（字符串，默认：
<没有>) - 集合表达式
-
用于评估MongoDB集合的SpEL表达式。（表达式，默认：
<没有>)
spring.data.mongodb
- 额外主持人
-
额外的服务器主机。如果没有指定“host”，则无法用URI设置。其他主机将使用默认的Mongo端口27017。如果你想用不同的端口，可以用“host：port”语法。（List<String>，默认：
<没有>) - 认证数据库
-
认证数据库名称。（字符串，默认：
<没有>) - 自动索引创建
-
是否启用自动索引创建。（布尔值，默认：
<没有>) - 数据库
-
数据库名称。覆盖URI数据库。（字符串，默认：
<没有>) - 场命名策略
-
字段命名策略的完全限定名称。（职业<？>，默认：
<没有>) - 主机
-
Mongo服务器主机。无法用URI设置。（字符串，默认：
<没有>) - 密码
-
Mongo服务器的登录密码。无法用URI设置。（字符[]，默认：
<没有>) - 端口
-
Mongo服务器端口。无法用URI设置。（整数，默认：
<没有>) - 复制集名称
-
集群的副本集名称要求。无法用URI设置。（字符串，默认：
<没有>) - 乌里
-
Mongo数据库URI。覆盖主机、端口、用户名和密码。（字符串，默认：
mongodb://localhost/test) - 用户名
-
登录Mongo服务器的用户。无法用URI设置。（字符串，默认：
<没有>) - UUID 表示
-
将 UUID 转换为 BSON 二进制值时使用的表示方式。（UuidRepresentation，默认：
Java-遗产，可能的值：未具体说明,标准,C_SHARP_LEGACY,JAVA_LEGACY,PYTHON_LEGACY)
spring.data.mongodb.gridfs
- 桶
-
GridFS 桶名称。（字符串，默认：
<没有>) - 数据库
-
GridFS 数据库名称。（字符串，默认：
<没有>)
spring.data.mongodb.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<没有>) - 启用
-
是否启用SSL支持。除非另有说明，否则若提供了“捆绑”，则自动启用。（布尔值，默认：
<没有>)
7.10. MQTT 汇入
该模块向MQTT发送消息。
7.10.1. 有效载荷：
-
字节[] -
字符串
7.10.2. 选项
mqtt 汇具以下选项：
按前缀分组的属性：
MQTT
- 清洁录音
-
客户端和服务器是否应该在重启和重新连接时记住状态。（布尔值，默认：
true) - 连接超时
-
连接超时只需几秒钟。（整数，默认：
30) - 保持生命间隔
-
延迟以秒计。（整数，默认：
60) - 密码
-
连接经纪人时使用的密码。（字符串，默认：
客人) - 坚持
-
“记忆”或“文件”。（字符串，默认：
存储) - 持久性目录
-
持久化目录。（字符串，默认：
/tmp/帕霍) - SSL-性质
-
MQTT 客户端 SSL 属性。（Map<String， String>，默认：
<没有>) - 网址
-
MQTT经纪人的位置（逗号分隔列表）。（字符串[]，默认：
[tcp://localhost:1883]) - 用户名
-
连接经纪人时使用的用户名。（字符串，默认：
客人)
mqtt.consumer
- 异步
-
是否使用异步发送。（布尔值，默认：
false) - 字符集
-
用于将字符串有效载荷转换为字节[]的字符集。（字符串，默认：
UTF-8) - 客户端ID
-
识别客户。（字符串，默认：
Stream.client.id.sink) - QoS（服务质量）
-
服务质量。（整数，默认：
1) - 保留
-
是否设置“保留”旗帜。（布尔值，默认：
false) - 主题
-
《The Sink》将发表的主题。（字符串，默认：
Stream.mqtt)
7.11. Pgcopy Shang
一个模块，通过 PostgreSQL COPY 命令将其收到的有效载荷写入 RDBMS。
7.11.1. 输入
头
有效载荷
-
任何
列表达式会根据消息进行评估，表达式通常只兼容一种类型（如 Map 或 bean 等）。
7.11.2. 输出
无
7.11.3. 期权
jdbc 汇具备以下选项：
- Spring.datasource.driver-class-name
-
JDBC驱动的名称是完全限定的。默认情况下，基于URL自动检测。（字符串，默认：
<没有>) - spring.datasource.password（春.datasource.password）
-
数据库的登录密码。（字符串，默认：
<没有>) - spring.datasource.url
-
数据库的JDBC网址。（字符串，默认：
<没有>) - spring.datasource.username
-
数据库用户名登录。（字符串，默认：
<没有>)
该模块还使用 Spring Boot 的 DataSource 支持来配置数据库连接，因此有以下属性spring.datasource.url 等等，申请。 |
7.11.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps你可以在这里找到相应的基于活页夹的项目。然后你可以把光盘放进其中一个文件夹里，然后构建：
$ ./mvnw clean package为了运行集成测试，可以在localhost上启动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 沉没
该模块向RabbitMQ发送消息。
7.12.1. 期权
兔子水槽有以下选项：
（有关RabbitMQ连接属性，请参见Spring Boot文档）
按前缀分组的属性：
兔
- 转换器豆名
-
自定义消息转换器的豆子名称;如果省略，则使用 SimpleMessageConverter。如果是“jsonConverter”，则会为你创建一个 Jackson2JsonMessageConverter 豆。（字符串，默认：
<没有>) - 交换
-
交换名称——如果提供了，则被exchangeNameExpression覆盖。（字符串，默认：
<空字符串>) - 交换表达式
-
一个评估交换名称的SpEL表达式。（表达式，默认：
<没有>) - 头部映射最后
-
在映射出站消息的头部时，确定头部是在消息转换前映射，还是转换后。（布尔值，默认：
true) - 映射请求头
-
这些头部将被映射。（字符串[]，默认：
[*]) - 自有连接
-
如果是真的，根据启动属性使用单独的连接。（布尔值，默认：
false) - 持久传递模式
-
当没有“amqp_deliveryMode”头部时，默认传递模式，对于持久性（PERSISTENT）成立。（布尔值，默认：
false) - 路由密钥
-
路由密钥——如果提供了，将被 routingKeyExpression 覆盖。（字符串，默认：
<没有>) - 路由密钥表达式
-
一个计算到路由键的 SpEL 表达式。（表达式，默认：
<没有>)
Spring.rabbitmq
- 地址洗牌模式
-
用于洗牌配置地址的模式。（AddressShuffleMode，默认：
没有，可能的值：没有,随机,顺序) - 地址
-
客户端应连接的地址列表，以逗号分隔。设置后，主机和端口被忽略。（字符串，默认：
<没有>) - 通道-RPC超时
-
通道中RPC调用的续时间超时。将其设为零以永久等待。（持续时间，默认：
10米) - 连接超时
-
连接超时。设置为零以永远等待。（持续时间，默认：
<没有>) - 主机
-
RabbitMQ 主机。如果设置了地址，则忽略。（字符串，默认：
本地主持) - 密码
-
登录以验证经纪人身份。（字符串，默认：
客人) - 端口
-
RabbitMQ端口。如果设置了地址，则忽略。默认为5672，启用SSL则为5671。（整数，默认：
<没有>) - 出版者-确认-类型
-
出版商类型确认使用。（ConfirmType，默认：
<没有>，可能的值：简单,相关,没有) - 出版商回归
-
是否启用出版商退货。（布尔值，默认：
false) - 请求通道最大值
-
客户端请求的每个连接通道数。无限使用0。（整数，默认：
2047) - 请求心跳
-
请求心跳超时;无则为零。如果未指定持续时间后缀，则使用秒数。（持续时间，默认：
<没有>) - 用户名
-
登录用户以向经纪人进行身份验证。（字符串，默认：
客人) - 虚拟主机
-
连接经纪商时使用的虚拟主机。（字符串，默认：
<没有>)
spring.rabbitmq.模板
- 默认接收队列
-
当没有消息时，默认接收消息的队列名称。（字符串，默认：
<没有>) - 交换
-
用于发送作的默认交换机名称。（字符串，默认：
<空字符串>) - 命令的
-
是否启用强制性消息。（布尔值，默认：
<没有>) - 接收超时
-
receive（）作的超时。（持续时间，默认：
<没有>) - 回复-超时
-
sendAndReceive（）作的超时。（持续时间，默认：
<没有>) - 路由密钥
-
用于发送作的默认路由密钥值。（字符串，默认：
<空字符串>)
7.13. Redis沉没
给Redis发消息。
7.13.1. 期权
Redis 水槽有以下选项：
按前缀分组的属性：
redis.consumer
- 钥匙
-
存储到密钥时使用的字面键名。（字符串，默认：
<没有>) - key-expression
-
一个用于存储到键的 SpEL 表达式。（字符串，默认：
<没有>) - 队列
-
存储队列时的字面队列名称。（字符串，默认：
<没有>) - 队列表达式
-
一个用于队列的 SpEL 表达式。（字符串，默认：
<没有>) - 主题
-
一个用于发布主题时使用的字面主题名称。（字符串，默认：
<没有>) - 主题表达
-
一个用于主题的特殊英语表达式。（字符串，默认：
<没有>)
spring.data.redis
- 客户名
-
客户端名称应与客户端 SETNAME 连接时设置。（字符串，默认：
<没有>) - 客户端类型
-
客户端类型。默认情况下，根据类路径自动检测。（ClientType，默认：
<没有>，可能的值：生菜,杰迪斯) - 连接超时
-
连接超时。（持续时间，默认：
<没有>) - 数据库
-
连接工厂使用的数据库索引。（整数，默认：
0) - 主机
-
Redis服务器主机。（字符串，默认：
本地主持) - 密码
-
Redis 服务器的登录密码。（字符串，默认：
<没有>) - 端口
-
Redis服务器端口。（整数，默认：
6379) - 超时
-
读“暂停”。（持续时间，默认：
<没有>) - 网址
-
连接URL。覆盖主机、端口、用户名和密码。示例：redis://user：[email protected]：6379（字符串，默认：
<没有>) - 用户名
-
登录Redis服务器的用户名。（字符串，默认：
<没有>)
spring.data.redis.cluster
- 最大重定向
-
执行集群命令时应执行的最大重定向次数。（整数，默认：
<没有>) - 节点
-
用逗号分隔的“host：port”对列表，用于引导。这代表一个“初始”集群节点列表，且至少需要一个条目。（List<String>，默认：
<没有>)
spring.data.redis.jedis.pool
- 启用
-
是否启用该池。如果有“commons-pool2”可用，则自动启用。在Jedis中，在哨兵模式下池化是隐式启用的，且该设置仅适用于单节点设置。（布尔值，默认：
<没有>) - 最大主动
-
池在特定时间内可分配的最大连接数。为无限制使用负值。（整数，默认：
8) - 最大怠速
-
池中“空闲”连接的最大数量。使用负值表示空闲连接数量无限。（整数，默认：
8) - 等一下
-
连接分配应在池耗尽后才触发异常的最长阻塞时间。使用负值无限期阻塞。（持续时间，默认：
-1毫秒) - 最小怠速
-
目标是池中最少需要维持的空闲连接数。该设置仅在该设置和驱逐运行间距均为正值时生效。（整数，默认：
0) - 驱逐间隔时间
-
空闲对象evictor线程运行间隔时间。当正时，空闲对象Evictor线程开始，否则不会执行空闲对象驱逐。（持续时间，默认：
<没有>)
spring.data.redis.lettuce.pool
- 启用
-
是否启用该池。如果有“commons-pool2”可用，则自动启用。在Jedis中，在哨兵模式下池化是隐式启用的，且该设置仅适用于单节点设置。（布尔值，默认：
<没有>) - 最大主动
-
池在特定时间内可分配的最大连接数。为无限制使用负值。（整数，默认：
8) - 最大怠速
-
池中“空闲”连接的最大数量。使用负值表示空闲连接数量无限。（整数，默认：
8) - 等一下
-
连接分配应在池耗尽后才触发异常的最长阻塞时间。使用负值无限期阻塞。（持续时间，默认：
-1毫秒) - 最小怠速
-
目标是池中最少需要维持的空闲连接数。该设置仅在该设置和驱逐运行间距均为正值时生效。（整数，默认：
0) - 驱逐间隔时间
-
空闲对象evictor线程运行间隔时间。当正时，空闲对象Evictor线程开始，否则不会执行空闲对象驱逐。（持续时间，默认：
<没有>)
spring.data.redis.lettuce
- 关机超时
-
关机暂停。（持续时间，默认：
100毫秒)
spring.data.redis.sentinel
- 主人
-
Redis 服务器名称。（字符串，默认：
<没有>) - 节点
-
“host：port”对的逗号分隔列表。（List<String>，默认：
<没有>) - 密码
-
用哨兵认证的密码。（字符串，默认：
<没有>) - 用户名
-
登录用户名用于用Sentinel认证。（字符串，默认：
<没有>)
spring.data.redis.ssl
- 捆
-
SSL 捆绑包名称。（字符串，默认：
<没有>) - 启用
-
是否启用SSL支持。除非另有说明，否则若提供了“捆绑”，则自动启用。（布尔值，默认：
<没有>)
7.14. 路由水槽
该应用程序将消息路由到指定信道。
7.14.1. 选项
路由器水槽有以下选项：
- router.default-output-binding
-
发送无法路由的消息。（字符串，默认：
<没有>) - router.destination-mappings
-
目的地映射作为新行分隔串的名称-值对，例如 'foo=bar\n baz=car'。（属性，默认：
<没有>) - router.expression
-
用于确定要路由到哪个通道的表达式。注意，文本、json或xml等内容类型的有效载荷线格式是字节[]，而非字符串！请参阅文档了解如何处理字节数组有效载荷内容。（表达式，默认：
<没有>) - router.refresh-delay
-
如果有，多久检查一次脚本变更（MS）;<0表示不要刷新。（整数，默认：
60000) - router.resolution-required（需要）
-
是否需要通道分辨率。（布尔值，默认：
false) - router.script
-
返回通道或通道映射分辨率键的groovy脚本位置。（资源，默认：
<没有>) - router.variables
-
变量绑定作为新行分隔串的名称-值对，例如 'foo=bar\n baz=car'。（属性，默认：
<没有>) - router.variables-location
-
包含自定义脚本变量绑定的属性文件的位置。（资源，默认：
<没有>)
这个路由器水槽基于流桥Spring Cloud Stream 的 API，因此可以根据需要创建目的地。
在这种情况下，adefaultOutputBinding只有当密钥不包含于 时才能访问目的地地图.
这必要分辨率 = 真忽略defaultOutputBinding如果没有映射和相应的绑定，则抛出异常。 |
你可以用spring.cloud.stream.dynamic目的地财产。
默认情况下，所有已解析的目的地都会被动态绑定;如果该属性有逗号分隔的目的地名称列表，则只有这些目标会被绑定。
如果消息最终解析到不在该列表中的目的地，则会被路由到defaultOutputBinding，也必须出现在列表中。
这目的地地图用于将评估结果映射到实际的目的地名称。
7.14.2. 基于特殊语言的路由
该表达式会根据消息进行评估，返回信道名称或信道名称映射的密钥。
欲了解更多信息，请参阅 Spring 集成参考手册“配置通用路由器”章节中的“路由器与 Spring 表达式语言（SpEL）”子章节。
7.14.3. 基于Groovy的路由
除了 SpEL 表达式，也可以使用 Groovy 脚本。我们在文件系统里创建一个Groovy脚本，地址为“file：/my/path/router.groovy”，或者“classpath：/my/path/router.groovy”：
println("Groovy processing payload '" + payload + "'")
if (payload.contains('a')) {
return "foo"
}
else {
return "bar"
}
如果你想把变量值传递给脚本，可以用变量选项静态绑定值，或者选择用propertiesLocation选项将路径传递到包含绑定的properties文件。 文件中的所有属性都将作为变量向脚本开放。你可以同时指定变量和 propertiesLocation，在这种情况下，任何作为变量提供的重复值都会覆盖 propertiesLocation 中的值。 注意，payload 和 header 是隐式绑定的，可以让你访问消息中包含的数据。
更多信息请参见 Spring Integration 参考手册《Groovy Support》。
7.15. 罗克特沉没
RSocket 汇入用 RSocket 协议的“发射后遗忘”策略发送数据。
7.15.1. 期权
Rsocket水槽有以下选项：
- rsocket.consumer.host
-
RSocket 主机。（字符串，默认：
本地主持) - rsocket.consumer.port
-
端口。（整数，默认：
7000) - rsocket.consumer.route
-
用于RSocket的路线。（字符串，默认：
<没有>) - rsocket.consumer.uri
-
可用于基于Websocket的传输的URI。（URI，默认：
<没有>)
7.16. 亚马逊S3沉没
这个汇入应用支持将对象转移到亚马逊S3桶。
文件、有效载荷（以及递归的目录）会被传输到远程目录（S3桶）映射到当地应用程序部署的目录。
该汇入接收的消息必须包含以下有效载荷如：
-
文件，包括递归上传目录; -
输入流; -
字节[]
7.16.1. 期权
S3水槽有以下选项：
按前缀分组的属性：
s3.consumer
- 前交叉口
-
S3 对象访问控制列表。（ObjectCannedACL，默认：
<没有>，可能的值：私人,公开阅读,公网-读-写,认证读取,aws-exec-read,桶所有者-读取,桶所有者-完全控制,零) - ACL表达式
-
用表达式来评估S3对象访问控制列表。（表达式，默认：
<没有>) - 桶
-
用于存储目标文件的 AWS 桶。（字符串，默认：
<没有>) - 桶表达式
-
用表达式来评估AWS的桶名称。（表达式，默认：
<没有>) - key-expression
-
用于评估S3对象键的表达式。（表达式，默认：
<没有>)
Spring.cloud.aws.credentials
- 访问密钥
-
访问密钥用于静态提供者。（字符串，默认：
<没有>) - 实例配置文件
-
无需额外配置即可配置实例配置文件凭证提供者。（布尔值，默认：
false) - 轮廓
-
AWS配置文件。（配置文件，默认：
<没有>) - 秘密密钥
-
与静态提供者一起使用的密钥。（字符串，默认：
<没有>)
spring.cloud.aws.region
- 实例配置文件
-
无需额外配置即可配置实例配置文件区域提供者。（布尔值，默认：
false) - 轮廓
-
AWS配置文件。（配置文件，默认：
<没有>) - 静态的
-
<缺少文档>（字符串，默认：
<没有>)
春云.AWS3
- 加速模式启用
-
在访问 S3 时启用加速端点的选项。加速端点通过利用亚马逊云前线的全球分布式边缘位置，实现了更快速的对象传输。（布尔值，默认：
<没有>) - 校验和验证启用
-
有个选项可以禁用对存储在 S3 中对象的校验和进行验证。（布尔值，默认：
<没有>) - 分块编码支持功能
-
在签署请求载荷时启用分块编码的选项，适用于{@link software.amazon.awssdk.services.s3.model.PutObjectRequest}和{@link software.amazon.awssdk.services.s3.model.UploadPartRequest}。（布尔值，默认：
<没有>) - 跨区域支持
-
支持跨区域桶访问。（布尔值，默认：
<没有>) - 端点
-
覆盖默认端点。（URI，默认：
<没有>) - 路径-样式-访问-启用
-
启用路径风格访问以访问S3对象，而非DNS风格访问的选项。偏好DNS风格的访问，因为这样在访问S3时能实现更好的负载均衡。（布尔值，默认：
<没有>) - 地区
-
覆盖默认区域。（字符串，默认：
<没有>) - use-arn-region-enabled
-
如果作为S3作的目标传递了一个与客户端配置区域不同的S3作目标，则该标志必须设置为“true”，以便客户端可以对ARN指定区域进行跨区域调用，否则将抛出异常。（布尔值，默认：
<没有>)
Spring.cloud.aws.s3.crt
- 初始读取缓冲区大小（字节）
-
配置客户端用来缓冲从S3下载的部件的起始缓冲区大小。保持更大的窗口以保持高下载吞吐量;除非窗口足够大能容纳多个零件，否则零件无法并行下载。保持较小的窗口以限制内存中缓冲的数据量。（长，默认：
<没有>) - 最大并发
-
规定传输过程中应建立的最大S3连接数。（整数，默认：
<没有>) - 最小部分大小（字节）
-
设定转移零件的最小零件尺寸。减小最小零件尺寸会导致多部件转移被拆分为更多小部件。设置过低会对传输速度产生负面影响，导致每个部分的延迟和网络通信增加。（长，默认：
<没有>) - 目标吞吐量（GBPS 中）
-
传输请求的目标吞吐量。价值越高，S3连接越多。传输管理器是否能实现配置的目标吞吐量取决于多种因素，如环境的网络带宽和配置的“最大并发”。（双重，默认：
<没有>)
目标生成的应用程序基于AmazonS3SinkConfiguration可以通过以下方式增强S3MessageHandler.UploadMetadataProvider和/或S3ProgressListener（进步听众），这些 被注入到S3MessageHandler豆。
详情请参见 Spring 集成 AWS 支持。
7.16.2. 亚马逊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.profilePath
其他的则是AWS的地区定义：
-
cloud.aws.region.auto
-
cloud.aws.region.static
例子
java -jar s3-sink.jar --s3.bucket=/tmp/bar7.17. SFTP沉没
SFTP sink 是一个简单的选项，可以将收到的消息推送文件到 SFTP 服务器。
它使用了一个SFTP-出站适配器因此，收到的消息可以是java.io.file对象，a字符串（文件内容）
或字节（文件内容也包括）
使用这个水槽需要用户名和密码才能登录。
默认情况下，Spring 集成会使用o.s.i.file.DefaultFileNameGenerator如果没有指定。DefaultFileNameGenerator将决定文件名
基于file_name在消息头，或者如果 的有效载荷消息已经是java.io.file那么它就会
使用该文件的原始名称。 |
在配置sftp.factory.known-hosts-expressionoption，评估的根对象是应用上下文，例如：sftp.factory.known-hosts-expression = @systemProperties['user.home'] + '/.ssh/known_hosts'.
7.17.1. 输入
头
-
file_name（见上文注释）
有效载荷
-
java.io.file -
java.io.InputStream -
字节[] -
字符串
7.17.2. 输出
不行（写入SFTP服务器）。
7.17.3. 期权
sftp 水槽有以下选项：
按前缀分组的属性：
SFTP.Consumer
- 自动-创建-导演
-
是否创建远程目录。（布尔值，默认：
true) - 文件名表达式
-
一个用于生成远程文件名的 SpEL 表达式。（字符串，默认：
<没有>) - 模式
-
如果远程文件已经存在，则需要采取相应的作。（FileExistsMode，默认：
<没有>，可能的值：附加,APPEND_NO_FLUSH,失败,忽视,取代,REPLACE_IF_MODIFIED) - 远程指令
-
远程FTP目录。（字符串，默认：
/) - 远程文件分隔器
-
远程文件分离器。（字符串，默认：
/) - 临时-远程-DIR
-
一个临时目录，如果“isUseTemporaryFilename（）”为真，文件将被写入。（字符串，默认：
/) - tmp-file-后缀
-
转移过程中使用的后缀。（字符串，默认：
.tmp) - 使用临时文件名
-
是否写入临时文件并重命名。（布尔值，默认：
true)
sftp.consumer.factory
- 允许未知密钥
-
允许未知或更改的密钥确实如此。（布尔值，默认：
false) - 缓存会话
-
缓存会话。（布尔值，默认：
<没有>) - 主机
-
服务器的主机名。（字符串，默认：
本地主持) - 已知宿主表达式
-
一个 SpEL 表达式，解析为已知主机文件的位置。（表达式，默认：
<没有>) - 暗语
-
用户私钥的密码短语。（字符串，默认：
<空字符串>) - 密码
-
连接服务器的密码。（字符串，默认：
<没有>) - 端口
-
服务器的移植。（整数，默认：
22) - 私钥
-
用户私钥的资源位置。（资源，默认：
<没有>) - 用户名
-
连接服务器的用户名。（字符串，默认：
<没有>)
7.18. TCP 吸收
该模块通过编码器向TCP写入消息。
TCP是一种流式传输协议，需要某种机制来在线路上对消息进行帧。有若干编码器 可用，默认是“CRLF”。
7.18.1. 期权
TCP Sink 有以下选项：
按前缀分组的属性：
TCP.consumer
- 字符集
-
从字节转换为字符串时使用的字符集。（字符串，默认：
UTF-8) - 关闭
-
是否在每条消息后关闭插槽。（布尔值，默认：
false) - 编码器
-
发送消息时使用的编码器。（编码，默认：
<没有>，可能的值：CRLF,如果,零,STXETX,生,第一语言,L2,L4) - 主机
-
这个水槽将连接到主机。（字符串，默认：
<没有>)
TCP
- 蔚来
-
是否使用蔚来。（布尔值，默认：
false) - 端口
-
值得倾听的端口;作系统选择端口时，0。（整数，默认：
1234) - 反向查找
-
对远程IP地址进行反向DNS查找;如果为假，消息头部中仅包含IP地址。（布尔值，默认：
false) - 套接字超时
-
当未接收到数据时，关闭套接字前的超时（毫秒）。（整数，默认：
120000) - 使用直接缓冲区
-
是否使用直接缓冲。（布尔值，默认：
false)
7.18.2. 可用编码器
- CRLF（默认）
-
文本以回车（0x0d）结束，随后换行（0x0a）
- 如果
-
文本以换行符结束（0x0a）
- 零
-
以空字节（0x00）结束的文本
- STXETX
-
文本前置STX（0x02），结尾为ETX（0x03）
- 生
-
无结构——客户端通过关闭套接字表示完整消息
- 第一语言
-
数据前有一个一字节（无符号）长度字段（支持最多255字节）
- L2
-
数据前有一个两字节（无符号）长度字段（最多216-1字节）
- L4
-
数据前有一个四字节（带符号）长度字段（最多231-1字节）
7.20。推特消息汇
从认证用户向指定用户发送直接消息。
需要一个 JSON POST 正文，并且内容类型将 首部设置为application/json.
| 当用户收到消息时，在24小时内最多可发送5条回复。 每收到一条消息，都会重置24小时窗口和分配的5条消息。 在24小时窗口内发送第6条消息或在24小时窗口外发送消息将计入速率限制。 这种行为只在使用 POST direct_messages/事件/新端点时出现。 |
SpEL表达式用于计算输入消息的请求参数。
7.20.1. 期权
使用单引号（）来包裹'SpEL表达式属性。
例如，设置固定消息文本使用情况文本='固定文本'.
用于固定目标用户ID的使用。用户Id='666'. |
- 推特.消息.更新.media-id
-
一个媒体ID，用于与信息关联。一条私信只能引用一个媒体ID。（表达式，默认：
<没有>) - 推特.消息.更新.屏幕名
-
发送私信的用户的屏幕名。（表达式，默认：
<没有>) - 推特.消息.更新.文本
-
私信短信。根据需要进行URL编码。最大长度为10,000字符。（表达式，默认：
有效载荷) - twitter.message.update.user-id
-
发送私信的用户ID。（表达式，默认：
<没有>)
7.21. 推特更新汇报
更新认证用户当前的文本（例如推文）。
| 每次更新尝试时，更新文本都会与认证用户最近的推文进行比较。 任何可能导致重复的尝试都会被阻止，导致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. 期权
按前缀分组的属性：
推特更新
- 附件-URL
-
（特殊英语表达）为了避免某个URL被计入扩展推文的正文，请提供一个URL作为推文附件。该URL必须是推文永久链接，或直接消息深度链接。文本文本中必须保留任意的非Twitter网址。传递给attachment_url参数且与推文永久链接或私信深度链接不匹配的URL将在推文创建时失败并触发异常。（表达式，默认：
<没有>) - 显示坐标
-
（特殊英语表达）是否要标记一条推文发出的具体坐标。（表达式，默认：
<没有>) - in-reply-to-status-id
-
（特殊英语表达）更新时回复的文本ID是现有的。注意：除非该参数引用推文的作者在正文中被提及，否则该参数将被忽略。因此，您必须在更新中包含@username，其中用户名是该推文的作者。当 inReplyToStatusID 被设置时，auto_populate_reply_metadata也会自动被设置。之后确保@mentions的引导推文会从原始推文中被查找，并从这里添加到新推文中。随着回复链的增长，@mentions会附加到推文的元数据中，直到@mentions的上限被提升。如果原始推文已被删除，回复将失败。（表达式，默认：
<没有>) - 媒体标识
-
（SpEL表达式）一条逗号分隔的推文关联的media_ids列表。推文中最多可包含4张照片、1张动画GIF或1个视频。有关上传媒体的更多详情，请参见“上传媒体”。（表达式，默认：
<没有>) - 地点标识
-
（特种英语表达）世界上的一个位置。（表达式，默认：
<没有>) - 文本
-
（SpEL表达式）文本的文本更新。根据需要进行URL编码。t.co 链接环绕会影响字符数。默认为消息有效载荷（表达式，默认：
有效载荷)
推特.更新.位置
- 纬度
-
该推文所指地点的纬度。除非该参数处于-90.0至+90.0（含北为正）范围内，否则该参数将被忽略。如果没有对应的长参数，也会忽略。（表达式，默认：
<没有>) - 离子
-
这条推文所指地点的经度。经度的有效范围为-180.0至+180.0（东经为正），含。如果超出该范围、非数字、geo_enabled被禁用或无对应的纬度参数，该参数将被忽略。（表达式，默认：
<没有>)
7.22. 波前汇陷
Wavefront 汇接收消息 <？>，将其转换为 Wavefront 数据格式的指标，并直接发送给 Wavefront 或 Wavefront 代理。
支持常见的ETL用例，即需要清理、转换现有（历史）指标数据并存储在Wavefront中以便进一步分析。
7.22.1. 选项
Wavefront洗手台有以下选项：
- wavefront.api-token（Wavefront.API-token）
-
Wavefront API 访问Tokens。（字符串，默认：
<没有>) - 波前。度量表达式
-
一个对度量值进行评估的SpEL表达式。（表达式，默认：
<没有>) - wavefront.metric-name
-
指标名称。默认应用名称。（字符串，默认：
<没有>) - Wavefront.proxy-uri
-
Wavefront代理的网址。（字符串，默认：
<没有>) - 波前。来源
-
唯一的应用程序、主机、容器或实例，能够发出指标。（字符串，默认：
<没有>) - wavefront.tag-expression
-
与度量相关的自定义元数据集合。点标签不能为空。键的有效字符包括：字母数字、连字符（'-'）、下划线('_')、点（'.'）。对于值，允许使用任何字符，包括空格。要包含双引号，请用反斜线转义。反斜线不能是标签值的最后一个字符。点标签键和值组合的最大允许长度为254个字符（包括分隔键和值的'='为255个字符）。如果值更长，则将该点拒绝并记录（Map <String， Expression>，默认：
<没有>) - 波前。时间戳表达式
-
一个 SpEL 表达式，可以对度量的时间戳进行评估（可选）。（表达式，默认：
<没有>) - 波前线.uri
-
Wavefront环境的网址。（字符串，默认：
<没有>)
7.23. Websocket Sink
一个简单的Websocket Sink实现。
7.23.1. 期权
支持以下选项：
- websocket.consumer.log级
-
netty 通道的 logLevel 。默认为 <tt>WARN</tt>（字符串，默认：
<没有>) - 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：部署时间源
步骤3：部署Websocket-sink
最后启动一个websocket-sink跟踪模式，这样你就能看到由时间源日志中：
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. 执行器
有一个端点你可以用它来访问最后一个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. 输入
-
字节[]
7.24.2. 输出
有效载荷
无
7.24.3. 选项
zeromq 吸收具有以下选项：
按前缀分组的属性：
XMPP.Consumer
- 查特-托
-
XMPP的地址用于发送消息。（字符串，默认：
<没有>)
XMPP.工厂
- 主机
-
连接XMPP主机服务器。（字符串，默认：
<没有>) - 密码
-
连接用户的密码。（字符串，默认：
<没有>) - 端口
-
连接主机的端口。- 默认客户端端口：5222（整数，默认：
5222) - 资源
-
XMPP主机上的资源绑定。- 可以是空的，如果未设置服务器将生成一个（字符串，默认：
<没有>) - 安全模式
-
<缺少文档>（安全模式，默认：
<没有>，可能的值：必填,如果可能的话,禁用) - 服务名称
-
为XMPP域设置服务名称。（字符串，默认：
<没有>) - 订阅模式
-
<缺少文档>（订阅模式，默认：
<没有>，可能的值：accept_all,reject_all,手动) - 用户
-
连接用户应该连接。（字符串，默认：
<没有>)
7.24.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps你可以在这里找到相应的活页夹项目。 然后你可以把光盘放进某个文件夹里，然后构建它：
$ ./mvnw clean package7.25. ZeroMQ 汇入
“zeromq”接收器使消息能够发送到ZeroMQ套接字。
7.25.1. 输入
-
字节[]
7.25.2. 输出
有效载荷
无
7.25.3. 期权
zeromq 吸收具有以下选项：
- zeromq.consumer.connect-url
-
连接 ZeroMQ 套接字的连接URL。（字符串，默认：
<没有>) - zeromq.consumer.socket-type
-
连接应该建立的套筒类型。（SocketType，默认：
<没有>，可能的值：双,酒馆,子,要求,代表,贩子,路由器,拉,推,XPUB,XSUB,流,客户端,服务器,收音机,菜,渠道,同辈,生,散射,收集) - zeromq.consumer.topic
-
一个主题 SpEL 表达式，用于在向订阅者发送消息前评估主题。（表达式，默认：
<没有>)
7.25.4. 构建
$ ./mvnw clean install -PgenerateApps
$ cd apps你可以在这里找到相应的活页夹项目。 然后你可以把光盘放进某个文件夹里，然后构建它：
$ ./mvnw clean package7.25.5. 示例
java -jar zeromq-sink.jar --zeromq.consumer.connectUrl=tcp://server:port --zeromq.consumer.topic=