Spring Boot 环境下多 Kafka 集群配置与实战
1. 项目依赖管理
要在 Spring Boot 项目中启用消息队列功能，首先需要引入官方提供的 starter。请在构建脚本 pom.xml 中添加如下坐标：
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>2. 集群连接参数设置
针对单一环境或多集群场景，配置文件（推荐 application.yml）的写法有所不同。若需对接多个 Kafka 实例，建议采用自定义前缀进行隔离配置。
单集群模式示例
适用于标准单点部署，直接复用原生配置项：
server:
port: 8080
spring:
kafka:
bootstrap-servers: 192.168.1.100:9092
producer:
acks: 1
retries: 3
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: app-group-core
auto-offset-reset: earliest
enable-auto-commit: true
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer双集群隔离配置
为了区分不同业务线或测试/生产环境，可以通过扩展属性前缀来管理多个数据源。以下演示了"主集群"与"辅集群"的配置差异：
app:
messaging:
# 默认主库
primary-cluster-name: alpha
# Alpha 集群配置
clusters:
alpha:
bootstrap-servers: 10.0.0.1:9092
producer:
acks: 1
retries: 5
consumer:
group-id: consumer-alpha
beta:
bootstrap-servers: 10.0.0.2:9092
producer:
acks: -1
retries: 3
consumer:
group-id: consumer-beta3. Java 配置类封装
基于上述属性，我们需要在代码中通过 @Configuration 手动绑定 Bean。为了避免硬编码，利用 KafkaProperties 结合 @ConfigurationProperties 进行解耦。以下是重构后的配置类逻辑：
package com.enterprise.msg.kafka.config;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.util.ObjectUtils;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class MultiKafkaPropertiesConfig {
// 定义 Alpha 集群属性
@ConfigurationProperties(prefix = "app.messaging.clusters.alpha")
@Bean(name = "alphaKafkaProps")
public KafkaProperties alphaConfig() {
return new KafkaProperties();
}
// 定义 Beta 集群属性
@ConfigurationProperties(prefix = "app.messaging.clusters.beta")
@Bean(name = "betaKafkaProps")
public KafkaProperties betaConfig() {
return new KafkaProperties();
}
// —— 生产者模板 (Producer) ——
@Bean("alphaTemplate")
public KafkaTemplate<String, String> alphaTemplate(@Qualifier("alphaKafkaProps") KafkaProperties props) {
Map<String, Object> cfg = buildProducerMap(props);
return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(cfg));
}
@Bean("betaTemplate")
public KafkaTemplate<String, String> betaTemplate(@Qualifier("betaKafkaProps") KafkaProperties props) {
Map<String, Object> cfg = buildProducerMap(props);
return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(cfg));
}
private Map<String, Object> buildProducerMap(KafkaProperties props) {
Map<String, Object> map = new HashMap<>();
map.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getBootstrapServers());
map.put(ProducerConfig.RETRIES_CONFIG, props.getProducer().getRetries());
map.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
map.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
return map;
}
// —— 消费者工厂 (Consumer Factory) ——
@Bean("alphaListenerFactory")
public ConcurrentKafkaListenerContainerFactory<Object, Object> alphaListenerFactory(
@Qualifier("alphaKafkaProps") KafkaProperties props) {
return createListenerFactory(new DefaultKafkaConsumerFactory<>(props.buildConsumerProperties()));
}
@Bean("betaListenerFactory")
public ConcurrentKafkaListenerContainerFactory<Object, Object> betaListenerFactory(
@Qualifier("betaKafkaProps") KafkaProperties props) {
return createListenerFactory(new DefaultKafkaConsumerFactory<>(props.buildConsumerProperties()));
}
private ConcurrentKafkaListenerContainerFactory<Object, Object> createListenerFactory(ConsumerFactory<Object, Object> factory) {
ConcurrentKafkaListenerContainerFactory<Object, Object> container = new ConcurrentKafkaListenerContainerFactory<>();
container.setConsumerFactory(factory);
return container;
}
}同时，为了便于业务层调用，我们可以将具体的发送服务抽象出来：
package com.enterprise.msg.service;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
@Configuration
public class MessageQueueServiceRegistry {
public static abstract class BaseMessageService {
protected final KafkaTemplate<String, String> template;
public BaseMessageService(KafkaTemplate<String, String> t) { this.template = t; }
}
@Bean("mqAlphaService")
public BaseMessageService mqAlpha(@Qualifier("alphaTemplate") KafkaTemplate<String, String> tmpl) {
return new BaseMessageService(tmpl) {};
}
@Bean("mqBetaService")
public BaseMessageService mqBeta(@Qualifier("betaTemplate") KafkaTemplate<String, String> tmpl) {
return new BaseMessageService(tmpl) {};
}
}4. 业务接口与服务实现
定义通用的消息投递契约：
package com.enterprise.msg.api;
import org.springframework.util.concurrent.ListenableFuture;
public interface MQPublisher {
void dispatch(String targetTopic, String payloadJson);
}
对应实现类注入特定的 Template：
package com.enterprise.msg.impl;
import com.enterprise.msg.api.MQPublisher;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KakaPublishImpl implements MQPublisher {
private final KafkaTemplate<String, String> producer;
public KakaPublishImpl(KafkaTemplate<String, String> p) {
this.producer = p;
}
@Override
public void dispatch(String targetTopic, String payloadJson) {
// 这里简化处理，实际生产中应返回 Future 以便监控回调
producer.send(targetTopic, payloadJson);
}
}5. 控制器层验证
创建一个 REST 接口用于触发消息发送，模拟业务操作：
package com.enterprise.msg.controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
import com.enterprise.msg.impl.KakaPublishImpl;
@RestController
public class DemoController {
// 假设已注入了三个不同的服务实例
private final KakaPublishImpl defaultSvc;
private final KakaPublishImpl alphaSvc;
private final KakaPublishImpl betaSvc;
public DemoController(KakaPublishImpl d, @Qualifier("mqAlphaService") KakaPublishImpl a, @Qualifier("mqBetaService") KakaPublishImpl b) {
this.defaultSvc = d;
this.alphaSvc = a;
this.betaSvc = b;
}
@PostMapping("/api/kafka/test")
public void triggerSend() {
String msg1 = "{\"type\":\"default\",\"content\":\"Primary Cluster Msg\"}";
String msg2 = "{\"type\":\"alpha\",\"content\":\"Secondary Alpha Msg\"}";
String msg3 = "{\"type\":\"beta\",\"content\":\"Secondary Beta Msg\"}";
defaultSvc.dispatch("test-topic", msg1);
alphaSvc.dispatch("test-topic", msg2);
betaSvc.dispatch("test-topic", msg3);
}
}6. 监听器接收端设计
消费者端需要指定对应的 ContainerFactory，以确保连接到正确的集群：
package com.enterprise.msg.listener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class UnifiedMessageHandler {
private static final Logger log = LoggerFactory.getLogger(UnifiedMessageHandler.class);
@KafkaListener(topics = "test-topic")
public void onDefaultMsg(ConsumerRecord<String, String> rec) {
log.info("[Default Group] Received offset {}: {}", rec.offset(), rec.value());
}
@KafkaListener(topics = "test-topic", containerFactory = "alphaListenerFactory")
public void onAlphaMsg(ConsumerRecord<String, String> rec) {
log.info("[Alpha Group] Processing data: {}", rec.value());
}
@KafkaListener(topics = "test-topic", containerFactory = "betaListenerFactory")
public void onBetaMsg(ConsumerRecord<String, String> rec) {
log.info("[Beta Group] Handling payload: {}", rec.value());
}
}7. 启动日志与结果校验
应用启动后，控制台会打印 Kafka 客户端的版本信息及消费者组加入状态。关键日志片段如下所示：
INFO c.e.msg.listener.UnifiedMessageHandler - [Default Group] Received offset 102: {"type":"default","content":"Primary Cluster Msg"}
INFO c.e.msg.listener.UnifiedMessageHandler - [Alpha Group] Processing data: {"type":"alpha","content":"Secondary Alpha Msg"}
INFO c.e.msg.listener.UnifiedMessageHandler - [Beta Group] Handling payload: {"type":"beta","content":"Secondary Beta Msg"}从日志可以看出，三个不同的消费者方法分别监听了同一主题下的消息，且各自独立处理。这验证了多数据源配置（即多 ConsumerFactory）已成功生效，实现了消息流的路由隔离。
