Spring Boot Kafka

Kafka的核心概念，以及如何保證高可用性、可靠性、順序性等問題這裏就不贅述了，本文只記錄操做示例。

依賴配置

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: message # consumer group 消費者羣組
      auto-offset-reset: latest # 當各分區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產生的該分區下的數據

KafkaProducerConfig

存儲Java類MessageForm對象,須要自行替換

@Configuration
public class KafkaProducerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        return props;
    }

    @Bean
    public ProducerFactory<String, MessageForm> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, MessageForm> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

KafkaConsumerConfig

配置類上須要@EnableKafka註釋才能在Spring託管Bean上檢測@KafkaListener註解。

@EnableKafka
@Configuration
public class KafkaConsumerConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

        return props;
    }

    @Bean
    public ConsumerFactory<String, MessageForm> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
                new JsonDeserializer<>(MessageForm.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, MessageForm> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, MessageForm> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        return factory;
    }
}

KafkaMessageListener

在KafkaConsumerConfig中咱們配置了ConsumerFactory和KafkaListenerContainerFactory。當這兩個Bean成功註冊到Spring IOC容器中後，咱們即可以使用@KafkaListener註解來監聽消息了。

@Component
public class KafkaMessageListener {

    @Resource
    private MessageService messageServiceImpl;

    @KafkaListener(topics="message",groupId = "consumer")
    public void listen(MessageForm messageForm){
        messageServiceImpl.message(messageForm);
    }
}