|
对于最新的稳定版本，请使用 spring-cloud-stream 4.2.0！ |
在 Kafka Streams Binder 中绑定可视化和控制
从版本 3.1.2 开始，Kafka Streams Binder 支持绑定可视化和控制。
仅支持的两个生命周期阶段是STOPPED和STARTED.
生命周期阶段PAUSED和RESUMED在 Kafka Streams 活页夹中不可用。
为了激活绑定可视化和控制，应用程序需要包含以下两个依赖项。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>如果您更喜欢使用 webflux，则可以包含spring-boot-starter-webflux而不是标准的 Web 依赖项。
此外，您还需要设置以下属性：
management.endpoints.web.exposure.include=bindings为了进一步说明此功能，让我们使用以下应用程序作为指南：
@SpringBootApplication
public class KafkaStreamsApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaStreamsApplication.class, args);
}
@Bean
public Consumer<KStream<String, String>> consumer() {
return s -> s.foreach((key, value) -> System.out.println(value));
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> function() {
return ks -> ks;
}
}正如我们所看到的，该应用程序有两个 Kafka Streams 函数 - 一个是使用者，另一个是函数。
默认情况下，使用者绑定命名为consumer-in-0.
同样，对于该函数，输入绑定为function-in-0输出绑定为function-out-0.
应用程序启动后，我们可以使用以下 bindings 端点查找有关绑定的详细信息。
curl http://localhost:8080/actuator/bindings | jq .
[
{
"bindingName": "consumer-in-0",
"name": "consumer-in-0",
"group": "consumer-applicationId",
"pausable": false,
"state": "running",
"paused": false,
"input": true,
"extendedInfo": {}
},
{
"bindingName": "function-in-0",
"name": "function-in-0",
"group": "function-applicationId",
"pausable": false,
"state": "running",
"paused": false,
"input": true,
"extendedInfo": {}
},
{
"bindingName": "function-out-0",
"name": "function-out-0",
"group": "function-applicationId",
"pausable": false,
"state": "running",
"paused": false,
"input": false,
"extendedInfo": {}
}
]有关所有三个绑定的详细信息，请参见上面。
现在让我们停止 consumer-in-0 绑定。
curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/consumer-in-0此时，不会通过此绑定接收任何记录。
再次启动绑定。
curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/consumer-in-0当单个函数上存在多个绑定时，对这些绑定中的任何一个调用这些作都将起作用。
这是因为单个函数上的所有绑定都由相同的StreamsBuilderFactoryBean.
因此，对于上述函数，要么function-in-0或function-out-0将起作用。