Spring Kafka多集群场景下如何为消费者指标添加自定义静态标签？

方案1：自定义KafkaListener标签生成器 #

实现KafkaListenerTagsProvider接口，替换默认的标签生成逻辑，直接给listener指标追加自定义标签：

@Component
public class CustomKafkaListenerTagsProvider implements KafkaListenerTagsProvider {

    @Value("${spring.kafka.cluster.name}")
    private String clusterName;

    @Override
    public Iterable<Tag> consumerTags(Consumer<?, ?> consumer, MessageListenerContainer container) {
        return Arrays.asList(
            Tag.of("cluster_name", clusterName),
            Tag.of("spring.id", container.getListenerId())
            // 可根据需求添加更多静态标签，比如topic名称、环境标识等
        );
    }
}

Spring会自动识别这个自定义实现，替代默认的DefaultKafkaListenerTagsProvider，所有Kafka listener相关的指标都会带上你定义的标签。

方案2：手动绑定指标时追加标签 #

如果需要更灵活的控制（比如不同容器对应不同集群），可以通过KafkaListenerEndpointRegistry获取容器实例，手动给指标添加标签：

@Autowired
private KafkaListenerEndpointRegistry registry;

@Autowired
private MeterRegistry meterRegistry;

@PostConstruct
public void registerCustomMetrics() {
    registry.getListenerContainers().forEach(container -> {
        // 根据容器获取对应的集群名称，比如从容器的KafkaConsumerFactory中提取配置
        String clusterName = getClusterNameFromContainer(container);
        
        // 给消费总数指标添加标签
        meterRegistry.gauge("kafka.listener.records.consumed.total",
            Tags.of("cluster_name", clusterName, "spring.id", container.getListenerId()),
            container,
            c -> c.getRecordMetrics().getConsumedCount());
        
        // 同理可给延迟、错误数等其他指标添加标签
        meterRegistry.gauge("kafka.listener.records.consumed.latency",
            Tags.of("cluster_name", clusterName, "spring.id", container.getListenerId()),
            container,
            c -> c.getRecordMetrics().getAvgLatency());
    });
}

// 自行实现从容器中提取集群名称的逻辑
private String getClusterNameFromContainer(MessageListenerContainer container) {
    // 示例：从consumer factory的配置中获取bootstrap servers，解析为集群名称
    ConsumerFactory<?, ?> factory = container.getConsumerFactory();
    String bootstrapServers = factory.getConfigurationProperties().getProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
    return mapBootstrapToClusterName(bootstrapServers); // 自定义映射逻辑
}

方案3：全局指标标签（应用级静态标签） #

如果集群名称是整个应用的静态标识（单应用只连接一个集群），可以给Micrometer注册全局通用标签，所有指标都会自动带上：

@Bean
public MeterRegistryCustomizer<MeterRegistry> commonTags() {
    return registry -> registry.config().commonTags(
        "cluster_name", "${spring.kafka.cluster.name}",
        "env", "${spring.profiles.active}"
    );
}

这种方式简单直接，但会给所有应用指标（包括非Kafka相关的）添加标签，适合不需要区分指标类型的场景。

注意：不同集群的应用实例要确保spring.kafka.cluster.name配置了对应的值，比如通过不同的配置文件或环境变量来区分。

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