SpringBoot Kafka动态指定Topic发送消息的实现方案咨询

1. 重构Producer类，支持动态传入Topic名称 #

移除Producer中依赖固定Topic的配置，修改send方法直接接收Topic名称参数：

public class Producer<Req> {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final KafkaTemplate<String, Req> template;

    public Producer(KafkaTemplate<String, Req> template) {
        this.template = template;
    }

    public void send(String topicName, Req requestStub) throws ExecutionException, InterruptedException, TimeoutException {
        logger.info("{}::send() to topic: {}", getClass().getCanonicalName(), topicName);

        final ProducerRecord<String, Req> record = new ProducerRecord<>(topicName, requestStub);
        ListenableFuture<SendResult<String, Req>> future = template.send(record);

        future.addCallback(new ListenableFutureCallback<SendResult<String, Req>>() {
            @Override
            public void onSuccess(SendResult<String, Req> result) {
                logger.info("Sent message=[{}] to topic=[{}] with offset=[{}]",
                        requestStub, topicName, result.getRecordMetadata().offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                logger.error("Unable to send message=[{}] to topic=[{}] due to: {}",
                        requestStub, topicName, ex.getMessage());
            }
        });
    }

    // 保留其他配置项（如超时），移除固定TopicName
    @Getter
    @Builder
    @ToString
    public static class Config {
        private final int sendTimeout;
    }
}

2. 配置Kafka管理客户端，实现动态创建Topic #

在配置类中添加KafkaAdmin和AdminClient Bean，用于检查和创建Topic：

@EnableKafka
@Configuration
class TimeOnTaskConfig {

    @Bean
    public Producer<TaskEvent> timeOnTaskProducer(final KafkaTemplate<String, TaskEvent> template) {
        return new Producer<>(template);
        // 若需配置超时等参数，可改为：
        // return new Producer<>(Producer.Config.builder().sendTimeout(5000).build(), template);
    }

    @Bean
    public KafkaTemplate<String, TaskEvent> kafkaTemplate(final ProducerFactory<String, TaskEvent> pf) {
        return new KafkaTemplate<>(pf);
    }

    @Bean
    public KafkaAdmin kafkaAdmin(KafkaProperties kafkaProperties) {
        KafkaAdmin admin = new KafkaAdmin(kafkaProperties.buildAdminProperties());
        admin.setFatalIfBrokerNotAvailable(true);
        return admin;
    }

    @Bean
    public AdminClient adminClient(KafkaAdmin kafkaAdmin) {
        return AdminClient.create(kafkaAdmin.getConfigurationProperties());
    }
}

3. 修改Service类，实现动态Topic逻辑 #

注入AdminClient，在发送消息前生成动态Topic名称，检查是否存在，不存在则创建，再调用Producer发送：

@Service
@Slf4j
public class TimeOnTaskService {

    private final Producer<TaskEvent> taskProducer;
    private final AdminClient adminClient;

    public TimeOnTaskService(final Producer<TaskEvent> taskProducer, AdminClient adminClient) {
        this.taskProducer = taskProducer;
        this.adminClient = adminClient;
    }

    public void enqueue(final TaskEvent event) {
        try {
            String topicName = "section" + event.getID();
            createTopicIfNotExists(topicName);
            taskProducer.send(topicName, event);
        } catch (Exception e) {
            log.error("enqueue method failed in TimeOnTaskService. {}", e.getMessage());
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    private void createTopicIfNotExists(String topicName) throws ExecutionException, InterruptedException {
        DescribeTopicsResult topicsResult = adminClient.describeTopics(Collections.singletonList(topicName));
        Map<String, TopicDescription> topicDescriptions = topicsResult.all().get();
        
        if (!topicDescriptions.containsKey(topicName)) {
            NewTopic newTopic = TopicBuilder.name(topicName)
                    .partitions(1)
                    .replicas(1)
                    .build();
            CreateTopicsResult createResult = adminClient.createTopics(Collections.singletonList(newTopic));
            createResult.all().get();
            log.info("Created new topic: {}", topicName);
        }
    }
}

关键说明 #

• 手动创建Topic比依赖Kafka的auto.create.topics.enable配置更可控，可自定义分区数、副本数等参数
• 使用AdminClient的同步方法（get()）确保Topic创建完成后再发送消息，避免消息发送失败
• 若需全局统一分区数/副本数，可将配置项写入application.yml并注入使用

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