在Spring Cloud Kafka中，實現消息廣播（也稱為發布-訂閱模式）主要依賴于Kafka的KafkaTemplate和Consumer。以下是實現消息廣播的步驟：
配置Kafka：
首先，確保你的Spring Boot應用程序已經正確配置了Kafka。你可以在application.yml或application.properties文件中添加Kafka配置。
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
group-id: my-group
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
創建Kafka生產者：
使用KafkaTemplate來發送消息。你可以創建一個配置類來初始化KafkaTemplate。
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaProducerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
發送廣播消息：
使用KafkaTemplate發送消息時，不需要指定分區鍵，因為Kafka會自動將消息廣播到所有分區的主題。
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaMessageSender {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void sendBroadcastMessage(String topic, String message) {
kafkaTemplate.send(topic, message);
}
}
創建Kafka消費者： 創建一個消費者來消費廣播消息。由于是廣播模式，所有消費者都會收到相同的消息。
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistrar;
import org.springframework.kafka.config.MethodKafkaListenerEndpoint;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.config.MethodKafkaListenerEndpointRegistry;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaConsumerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
ErrorHandlingDeserializer<String> errorHandlingDeserializer = new ErrorHandlingDeserializer<>(new JsonDeserializer<>());
errorHandlingDeserializer.setFallbackToNull(true);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, errorHandlingDeserializer);
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
@Bean
public KafkaListenerEndpointRegistrar kafkaListenerEndpointRegistrar() {
return new KafkaListenerEndpointRegistrar();
}
@Bean
public MethodKafkaListenerEndpointRegistry methodKafkaListenerEndpointRegistry(KafkaListenerEndpointRegistrar registrar) {
MethodKafkaListenerEndpointRegistry registry = new MethodKafkaListenerEndpointRegistry();
registry.setEndpoints(kafkaListenerEndpoints());
registrar.registerEndpoints(registry);
return registry;
}
private Map<String, MethodKafkaListenerEndpoint<?>> kafkaListenerEndpoints() {
Map<String, MethodKafkaListenerEndpoint<?>> endpoints = new HashMap<>();
// 添加你的消費者方法
return endpoints;
}
}
創建消費者監聽器： 創建一個消費者監聽器來處理接收到的消息。
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class KafkaMessageListener {
@KafkaListener(topics = "${spring.kafka.consumer.topic}", groupId = "${spring.kafka.consumer.group-id}")
public void listen(String message) {
System.out.println("Received message: " + message);
}
}
通過以上步驟，你就可以在Spring Cloud Kafka中實現消息廣播了。生產者發送的消息會被廣播到指定的主題，所有訂閱該主題的消費者都會收到消息。
