SpringBoot自定义Kinesis健康指示器获取绑定目标流方法

为SpringBoot应用自定义KinesisBinderHealthIndicator，要求当配置文件中spring.cloud.stream.bindings下声明的目标Kinesis流，与AWS侧实际存在的流不匹配时（流被误删、未自动创建等场景），/actuator/health端点返回DOWN状态。

示例配置：

spring.cloud.stream.bindings.my-first-stream-in-0.destination=my-first-stream
spring.cloud.stream.bindings.my-second-stream-in-0.destination=my-second-stream

示例AWS流查询结果：

aws --endpoint-url=http://localhost:4566 kinesis list-streams
{
    "StreamNames": [
        "my-first-stream"
    ]
}

不需要手动读取配置文件拼接路径，直接注入Spring Cloud Stream自动装配的配置绑定类即可，步骤如下：

• 注入BindingsProperties Bean，这个类是框架原生用来承载所有spring.cloud.stream.bindings前缀配置的，自动适配properties/yml/配置中心/环境变量等所有配置源，支持宽松绑定规则，不会出现配置遗漏。
• 遍历所有binding配置项，提取每个配置的destination值做去重；如果应用同时使用多个binder（比如同时接入Kinesis、Kafka），可以额外通过BindingProperties#getBinder()过滤出归属Kinesis的配置，避免误统计其他中间件的目标资源。
• 对比配置声明的流列表和AWS实际查询到的流列表，若存在配置声明但实际缺失的流，返回DOWN状态，同时把缺失流信息写入健康检查详情方便排查。

补全后的完整代码 #

@Primary
@Component("kinesisBinderHealthIndicator")
@ComponentScan(basePackages = "org.springframework.cloud.stream.binder.kinesis")
@RequiredArgsConstructor
public class CustomKinesisBinderHealthIndicator implements HealthIndicator {

    private final KinesisMessageChannelBinder kinesisMessageChannelBinder;
    private final KinesisExtendedBindingProperties kinesisExtendedBindingProperties;
    // 注入Spring Cloud Stream全量binding配置
    private final BindingsProperties bindingsProperties;

    @Override
    public Health health() {
        try {
            // 这里替换成你实际查询AWS Kinesis现存流列表的逻辑
            List<String> actualKinesisStreams = new ArrayList<>(this.kinesisMessageChannelBinder.getStreamsInUse());
            Set<String> actualStreamSet = new HashSet<>(actualKinesisStreams);

            // 从配置中提取所有声明的目标流
            Set<String> configuredStreams = bindingsProperties.getBindings().values()
                    .stream()
                    // 多binder场景打开下面这行，替换为你实际使用的kinesis binder名称
                    // .filter(bindingProp -> "kinesis".equals(bindingProp.getBinder()))
                    .map(BindingProperties::getDestination)
                    .filter(Objects::nonNull)
                    .filter(dest -> !dest.isBlank())
                    .collect(Collectors.toSet());

            // 计算配置存在但实际缺失的流
            Set<String> missingStreams = configuredStreams.stream()
                    .filter(dest -> !actualStreamSet.contains(dest))
                    .collect(Collectors.toSet());

            if (!missingStreams.isEmpty()) {
                return Health.down()
                        .withDetail("missing_kinesis_streams", missingStreams)
                        .withDetail("configured_streams", configuredStreams)
                        .withDetail("actual_available_streams", actualStreamSet)
                        .build();
            }

            // 校验通过返回UP状态
            return Health.up()
                    .withDetail("available_streams", actualStreamSet)
                    .build();
        } catch (Exception e) {
            // 修复原代码多写右括号的语法问题
            return Health.down(e).build();
        }
    }
}

注意事项 #

• 单binder场景不需要加binder过滤逻辑，所有binding默认使用上下文里的唯一binder。
• 去重逻辑必须保留，因为多个binding（比如输入、输出绑定）可以指向同一个destination流，不需要重复统计。
• 不推荐直接通过Environment手动拼接spring.cloud.stream.bindings.*路径读配置，这种方式无法适配配置覆盖、宽松绑定等场景，容易出现漏读。
• 有特殊流过滤需求（比如排除测试临时流），可以直接在提取配置流的stream流程里加自定义过滤规则。

内容的提问来源于stack exchange，提问作者Lucian Radu