如何在Spring Boot的RESTful API中按需处理多Kafka主题？

针对你提出的需求——访问/requester1/data时从topic1取数，访问/requester2/data时从topic2取数这类场景，结合Spring Kafka的特性，我整理了两种落地性很强的实现方案，你可以根据自己的业务场景选择：

方案一：静态绑定多消费者（适合固定主题场景） #

如果你的主题数量是提前确定、不会频繁新增的，这种方案最简单直接——为每个主题预先配置独立的消费者，请求过来时直接调用对应逻辑拿数据。

1. 基础Kafka配置 #

先在application.yml里配置好消费者的基础属性：

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      group-id: api-data-consumer-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    properties:
      max.poll.records: 100  # 每次拉取的最大记录数

2. 为每个主题编写消费者 #

用@KafkaListener注解绑定主题，同时把消费到的数据存在线程安全的队列里，供接口调用：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Component
public class TopicDataConsumer {
    // 存储各主题的数据，按需扩展
    private final BlockingQueue<String> topic1Queue = new LinkedBlockingQueue<>();
    private final BlockingQueue<String> topic2Queue = new LinkedBlockingQueue<>();

    @KafkaListener(topics = "topic1", groupId = "api-data-consumer-group")
    public void consumeTopic1(String message) {
        topic1Queue.offer(message);
    }

    @KafkaListener(topics = "topic2", groupId = "api-data-consumer-group")
    public void consumeTopic2(String message) {
        topic2Queue.offer(message);
    }

    // 对外提供获取数据的方法，支持超时等待
    public String getTopic1Data() throws InterruptedException {
        return topic1Queue.poll(5, TimeUnit.SECONDS);
    }

    public String getTopic2Data() throws InterruptedException {
        return topic2Queue.poll(5, TimeUnit.SECONDS);
    }
}

3. 编写REST接口映射请求 #

在Controller里根据请求路径的requester参数，调用对应主题的取数方法：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.TimeUnit;

@RestController
public class DataController {
    private final TopicDataConsumer dataConsumer;

    // 构造注入，避免Autowired
    public DataController(TopicDataConsumer dataConsumer) {
        this.dataConsumer = dataConsumer;
    }

    @GetMapping("/{requester}/data")
    public String getData(@PathVariable String requester) throws InterruptedException {
        return switch (requester) {
            case "requester1" -> dataConsumer.getTopic1Data();
            case "requester2" -> dataConsumer.getTopic2Data();
            // 新增请求方直接在这里加case就行
            default -> "Sorry, we don't support this requester yet: " + requester;
        };
    }
}

方案二：动态创建消费者（适合主题动态扩展场景） #

如果你的主题数量不确定，需要随时新增，那动态创建消费者的方案更灵活——根据请求参数动态订阅对应主题，拉取数据后再销毁（或者复用）。

1. 配置消费者工厂 #

先配置好ConsumerFactory，用来动态生成Kafka消费者：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaDynamicConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "dynamic-api-consumer-group");
        configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(configProps);
    }
}

2. 实现动态拉取逻辑 #

利用KafkaConsumer手动创建消费者，订阅指定主题并拉取数据：

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

@Component
public class DynamicTopicFetcher {
    private final Properties consumerProps;

    public DynamicTopicFetcher(ConsumerFactory<String, String> consumerFactory) {
        this.consumerProps = new Properties();
        this.consumerProps.putAll(consumerFactory.getConfigurationProperties());
    }

    public String fetchData(String topic) {
        // 用try-with-resources自动关闭消费者
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            consumer.subscribe(Collections.singletonList(topic));
            // 拉取5秒内的最新数据，可根据需求调整时长
            var records = consumer.poll(Duration.ofSeconds(5));
            if (!records.isEmpty()) {
                return records.iterator().next().value();
            }
            return "No data available in topic: " + topic;
        }
    }
}

3. 编写动态映射的REST接口 #

维护一个请求方和主题的映射表（可以从配置文件读取，更灵活），然后根据请求参数匹配主题：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@RestController
public class DynamicDataController {
    private final DynamicTopicFetcher topicFetcher;
    // 请求方与主题的映射表，可配置化
    private final Map<String, String> requesterTopicMap = new ConcurrentHashMap<>();

    public DynamicDataController(DynamicTopicFetcher topicFetcher) {
        this.topicFetcher = topicFetcher;
        // 初始化映射，后续可以从配置中心或数据库读取
        requesterTopicMap.put("requester1", "topic1");
        requesterTopicMap.put("requester2", "topic2");
    }

    @GetMapping("/{requester}/data")
    public String getData(@PathVariable String requester) {
        String topic = requesterTopicMap.get(requester);
        if (topic == null) {
            return "Unsupported requester: " + requester;
        }
        return topicFetcher.fetchData(topic);
    }
}

方案对比与注意点 #

• 方案一：实现简单，性能稳定，适合主题固定的场景；缺点是新增主题需要改代码重启服务。
• 方案二：灵活性拉满，支持动态新增主题；缺点是每次创建消费者有一定开销，可以考虑加个消费者池复用优化。
• 注意group-id的配置，同一个组的消费者会分摊消费，不同组则会重复消费，根据你的业务需求选择。
• 如果需要批量获取数据，只要调整poll的参数或者返回集合类型就行。

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