Spring Boot中Kafka消费者动态主题订阅的实现方案咨询

问题背景 #

现有Kafka消费者工厂实现中，消费者启动时仅订阅一次主题，无法自动根据当前小时切换到对应的主题（如user_behaviour_EMAIL_9→user_behaviour_EMAIL_10）。需要改造实现每小时自动切换订阅主题的功能。

现有代码 #

ConsumerFactory 类 #

@Service
public class ConsumerFactory {
    private TopicNameProvider topicNameProvider;

    private final static boolean ENABLE_AUTO_COMMIT = false;
    private final static int MAX_POLL_RECORDS = 10;
    private final static int MAX_POLL_INTERVAL_MS = 600000;  // 10 min

    public ConsumerFactory(@Autowired(required = false) TopicNameProvider topicNameProvider) {
        this.topicNameProvider = topicNameProvider;
    }

    protected Properties fromKafkaProperties(KafkaConsumerProperties kafkaConsumerProperties){
        Properties props = new Properties();
        props.put("bootstrap.servers", kafkaConsumerProperties.getBootstrapServers());
        props.put("group.id", kafkaConsumerProperties.getGroupId());
        props.put("key.deserializer", ByteArrayDeserializer.class.getName());
        props.put("value.deserializer", ByteArrayDeserializer.class.getName());
        props.put("auto.offset.reset", "earliest");
        props.put("enable.auto.commit", String.valueOf(ENABLE_AUTO_COMMIT));
        props.put("max.poll.records", String.valueOf(MAX_POLL_RECORDS));
        props.put("max.poll.interval.ms", String.valueOf(MAX_POLL_INTERVAL_MS));
        return props;
    }

    @Bean
    public KafkaConsumer consumer(KafkaConsumerProperties kafkaConsumerProperties) {
        KafkaConsumer kafkaConsumer = new KafkaConsumer(fromKafkaProperties(kafkaConsumerProperties));
        kafkaConsumer.subscribe(topicNamesToSubscribe(kafkaConsumerProperties));
        return kafkaConsumer;
    }

    protected List<String> topicNamesToSubscribe(KafkaConsumerProperties kafkaConsumerProperties) {
        if (topicNameProvider != null) {
            return topicNameProvider.topicNames();
        } else if (kafkaConsumerProperties.getTopics() != null) {
            return Arrays.asList(kafkaConsumerProperties.getTopics());
        } else {
            throw new RuntimeException("No topic definition found to subscribe. Define topics either in application.yml " +
                    "in kafka.topics or implement a TopicsNameProvider bean.");
        }
    }
}

UserBehaviourTopicNameProvider 类 #

@Service
@ConditionalOnProperty(name = "messaging.mode", havingValue = "behaviour")
public class UserBehaviourTopicNameProvider implements TopicNameProvider {

    @Override
    public List<String> topicNames() {
        return List.of("user_behaviour_EMAIL_" + new Date().getHours());
    }
}

问题分析 #

当前实现的核心问题是：KafkaConsumer被注册为Spring单例Bean，启动时仅执行一次subscribe操作。即使topicNames()方法返回的主题随时间变化，消费者也不会主动重新订阅新主题。

解决方案 #

核心思路 #

1. 动态管理消费者实例：不再将KafkaConsumer注册为单例Bean，而是通过管理器类动态创建、销毁消费者实例。
2. 定时触发主题切换：使用定时任务，在每小时整点触发消费者切换逻辑，关闭旧消费者并创建新消费者订阅当前小时的主题。
3. 保证消费平滑切换：在切换过程中确保旧消费者完成当前批次的消息处理，避免消息丢失或重复消费。
4. 优化主题生成逻辑：缓存当前小时的主题名称，避免频繁重复计算。

具体实现代码 #

1. 改造ConsumerFactory，移除@Bean注册，提供创建消费者的方法

@Service
public class ConsumerFactory {
    private TopicNameProvider topicNameProvider;

    private final static boolean ENABLE_AUTO_COMMIT = false;
    private final static int MAX_POLL_RECORDS = 10;
    private final static int MAX_POLL_INTERVAL_MS = 600000;  // 10 min

    public ConsumerFactory(@Autowired(required = false) TopicNameProvider topicNameProvider) {
        this.topicNameProvider = topicNameProvider;
    }

    protected Properties fromKafkaProperties(KafkaConsumerProperties kafkaConsumerProperties){
        Properties props = new Properties();
        props.put("bootstrap.servers", kafkaConsumerProperties.getBootstrapServers());
        props.put("group.id", kafkaConsumerProperties.getGroupId());
        props.put("key.deserializer", ByteArrayDeserializer.class.getName());
        props.put("value.deserializer", ByteArrayDeserializer.class.getName());
        props.put("auto.offset.reset", "earliest");
        props.put("enable.auto.commit", String.valueOf(ENABLE_AUTO_COMMIT));
        props.put("max.poll.records", String.valueOf(MAX_POLL_RECORDS));
        props.put("max.poll.interval.ms", String.valueOf(MAX_POLL_INTERVAL_MS));
        return props;
    }

    // 提供创建消费者的方法，不再注册为Bean
    public KafkaConsumer<String, byte[]> createConsumer(KafkaConsumerProperties kafkaConsumerProperties) {
        KafkaConsumer<String, byte[]> kafkaConsumer = new KafkaConsumer<>(fromKafkaProperties(kafkaConsumerProperties));
        kafkaConsumer.subscribe(topicNamesToSubscribe(kafkaConsumerProperties));
        return kafkaConsumer;
    }

    protected List<String> topicNamesToSubscribe(KafkaConsumerProperties kafkaConsumerProperties) {
        if (topicNameProvider != null) {
            return topicNameProvider.topicNames();
        } else if (kafkaConsumerProperties.getTopics() != null) {
            return Arrays.asList(kafkaConsumerProperties.getTopics());
        } else {
            throw new RuntimeException("No topic definition found to subscribe. Define topics either in application.yml " +
                    "in kafka.topics or implement a TopicsNameProvider bean.");
        }
    }
}

2. 创建消费者管理器，处理定时切换逻辑

@Service
public class KafkaConsumerManager {
    private final ConsumerFactory consumerFactory;
    private final KafkaConsumerProperties kafkaConsumerProperties;
    private volatile KafkaConsumer<String, byte[]> currentConsumer;
    private final Object consumerLock = new Object();
    private ScheduledExecutorService scheduler;

    public KafkaConsumerManager(ConsumerFactory consumerFactory, KafkaConsumerProperties kafkaConsumerProperties) {
        this.consumerFactory = consumerFactory;
        this.kafkaConsumerProperties = kafkaConsumerProperties;
        // 初始化第一个消费者
        this.currentConsumer = consumerFactory.createConsumer(kafkaConsumerProperties);
        // 启动消费线程
        startConsuming();
        // 初始化定时任务，整点切换消费者
        initScheduler();
    }

    private void initScheduler() {
        scheduler = Executors.newSingleThreadScheduledExecutor();
        // 计算距离下一个整点的延迟时间
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime nextHour = now.plusHours(1).withMinute(0).withSecond(0).withNano(0);
        long initialDelay = Duration.between(now, nextHour).toMillis();
        // 每小时执行一次切换
        scheduler.scheduleAtFixedRate(this::switchConsumer, initialDelay, 3600000, TimeUnit.MILLISECONDS);
    }

    private void switchConsumer() {
        synchronized (consumerLock) {
            // 关闭旧消费者，确保提交已处理的偏移量
            if (currentConsumer != null) {
                try {
                    currentConsumer.commitSync();
                    currentConsumer.close(Duration.ofSeconds(10));
                } catch (Exception e) {
                    // 处理关闭异常
                    e.printStackTrace();
                }
            }
            // 创建新消费者并启动消费
            currentConsumer = consumerFactory.createConsumer(kafkaConsumerProperties);
        }
    }

    private void startConsuming() {
        new Thread(() -> {
            while (true) {
                KafkaConsumer<String, byte[]> consumer;
                synchronized (consumerLock) {
                    consumer = currentConsumer;
                }
                try {
                    ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
                    // 处理消息逻辑，替换为你的业务代码
                    for (ConsumerRecord<String, byte[]> record : records) {
                        // process record
                        System.out.println("Consumed record: " + record.topic() + " - " + record.offset());
                    }
                    // 手动提交偏移量
                    consumer.commitSync();
                } catch (WakeupException e) {
                    // 消费者被唤醒，通常是关闭时触发，无需处理
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    // 应用关闭时清理资源
    @PreDestroy
    public void shutdown() {
        scheduler.shutdown();
        synchronized (consumerLock) {
            if (currentConsumer != null) {
                currentConsumer.wakeup();
                currentConsumer.close();
            }
        }
    }
}

3. 优化主题名称生成逻辑（可选）

避免每次调用topicNames()都创建新的Date对象，缓存当前小时的主题：

@Service
@ConditionalOnProperty(name = "messaging.mode", havingValue = "behaviour")
public class UserBehaviourTopicNameProvider implements TopicNameProvider {
    private volatile String currentTopic;
    private volatile int lastHour = -1;

    @Override
    public List<String> topicNames() {
        int currentHour = LocalDateTime.now().getHour();
        if (currentHour != lastHour) {
            synchronized (this) {
                if (currentHour != lastHour) {
                    currentTopic = "user_behaviour_EMAIL_" + currentHour;
                    lastHour = currentHour;
                }
            }
        }
        return List.of(currentTopic);
    }
}

关键注意事项 #

• 线程安全：使用synchronized保证消费者切换时的线程安全，避免消费逻辑出现并发问题。
• 偏移量提交：切换消费者前必须手动提交偏移量，确保已处理的消息不会重复消费。
• 资源清理：关闭旧消费者时使用close(Duration)，给足够时间完成清理；应用关闭时触发shutdown()方法释放资源。
• 定时任务初始化：计算初始延迟确保第一次切换在整点触发，避免提前或延迟执行。

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