如何在spring-cloud-stream-kafka-binder中获取绑定的topic名称

实现方案 #

你可以通过 Spring Boot 配置绑定批量读取输出绑定配置，动态生成分支判断规则，无需硬编码每个分支的判断逻辑：

1. 定义配置绑定类

批量读取application.yml中所有spring.cloud.stream.bindings下的输出绑定配置：

import org.springframework.boot.context.properties.ConfigurationProperties;
import java.util.Map;

@ConfigurationProperties(prefix = "spring.cloud.stream.bindings")
public class StreamBindingProperties {
    // 自动映射所有 process-out-* 开头的绑定配置
    private Map<String, BindingConfig> processOut;

    public static class BindingConfig {
        private String destination;

        public String getDestination() {
            return destination;
        }
        public void setDestination(String destination) {
            this.destination = destination;
        }
    }

    public Map<String, BindingConfig> getProcessOut() {
        return processOut;
    }
    public void setProcessOut(Map<String, BindingConfig> processOut) {
        this.processOut = processOut;
    }
}

在启动类上添加注解开启配置绑定：

@EnableConfigurationProperties(StreamBindingProperties.class)
@SpringBootApplication
public class YourApplication {
    public static void main(String[] args) {
        SpringApplication.run(YourApplication.class, args);
    }
}

2. 动态生成分支规则

在处理逻辑中注入配置类，根据配置动态生成Predicate数组传入branch方法所需的判断规则：

import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Predicate;
import org.springframework.context.annotation.Bean;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;

public class WordCountProcessor {
    private final StreamBindingProperties bindingProperties;

    public WordCountProcessor(StreamBindingProperties bindingProperties) {
        this.bindingProperties = bindingProperties;
    }

    @Bean
    @SuppressWarnings("unchecked")
    public Function<KStream<Object, String>, KStream<?, WordCount>[]> process() {
        List<Predicate<Object, WordCount>> predicateList = new ArrayList<>();
        // 遍历所有输出绑定配置生成对应判断规则
        bindingProperties.getProcessOut().values().forEach(binding -> {
            // 这里根据你的业务规则映射destination和word的对应关系，示例中是把destination的前缀作为判断的word值
            String targetWord = binding.getDestination().replace("-counts", "");
            predicateList.add((k, v) -> v.getWord().equalsIgnoreCase(targetWord));
        });
        Predicate<Object, WordCount>[] predicates = predicateList.toArray(new Predicate[0]);

        return input -> input
                // 输入字符串转WordCount对象，可替换为你自己的转换逻辑
                .mapValues(wordStr -> new WordCount(wordStr, 1))
                .branch(predicates);
    }

    // WordCount实体类示例
    public static class WordCount {
        private String word;
        private long count;

        public WordCount(String word, long count) {
            this.word = word;
            this.count = count;
        }
        public String getWord() { return word; }
        public long getCount() { return count; }
    }
}

注意事项

• branch方法返回的KStream数组顺序和Predicate数组顺序完全一致，Spring Cloud Stream会自动将数组下标和process-out-0、process-out-1的输出绑定按顺序自动匹配
• 后续新增分支只需在application.yml中新增process-out-x配置即可，无需修改业务代码

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