Kafka 消息队列与 Java 微服务整合 (全网最完整教程)

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

spring:
  kafka:
    bootstrap-servers: localhost:9092 # Kafka集群地址
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 或 StringSerializer
      acks: all # 生产者发送消息的确认机制，all表示等待所有ISR副本确认
      retries: 3 # 重试次数
      batch-size: 16384 # 批量发送消息的大小，单位字节
      buffer-memory: 33554432 # 生产者可用于缓冲等待发送消息的总内存
    consumer:
      group-id: my-microservice-group # 消费者组ID
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer # 或 StringDeserializer
      auto-offset-reset: earliest # 首次启动或无offset时，从最早的offset开始消费
      enable-auto-commit: false # 关闭自动提交，手动控制提交时机
      max-poll-records: 500 # 每次poll操作最多拉取的消息数量

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class MessageProducer {

    private final KafkaTemplate<String, Object> kafkaTemplate; // 通常key是String，value可以是任何POJO，通过JsonSerializer序列化

    public MessageProducer(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String topic, String key, Object data) {
        // 异步发送消息
        kafkaTemplate.send(topic, key, data).addCallback(
            result -> System.out.println("发送成功: " + result.getProducerRecord().value()),
            ex -> System.err.println("发送失败: " + ex.getMessage())
        );
        // 如果需要同步发送，可以使用 .get() 方法，但不推荐，会阻塞
        // try {
        //     kafkaTemplate.send(topic, key, data).get();
        // } catch (Exception e) {
        //     e.printStackTrace();
        // }
    }
}

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class MessageConsumer {

    // 监听名为 "my-topic" 的主题，使用前面配置的消费者组ID
    @KafkaListener(topics = "my-topic", groupId = "my-microservice-group")
    public void listen(ConsumerRecord<String, Object> record, Acknowledgment ack) {
        System.out.println("收到消息 - Topic: " + record.topic() +
                           ", Key: " + record.key() +
                           ", Value: " + record.value() +
                           ", Offset: " + record.offset());
        // 处理消息的业务逻辑...
        // 模拟处理失败
        // if (Math.random() < 0.1) {
        //     throw new RuntimeException("模拟处理失败");
        // }

        // 手动提交offset，确保消息处理完成后才提交
        ack.acknowledge();
    }

    // 也可以批量消费消息
    // @KafkaListener(topics = "my-topic", groupId = "my-microservice-group", containerFactory = "kafkaListenerContainerFactory")
    // public void listenBatch(List<ConsumerRecord<String, Object>> records, Acknowledgment ack) {
    //     System.out.println("收到批量消息，数量: " + records.size());
    //     for (ConsumerRecord<String, Object> record : records) {
    //         // 处理单条消息
    //     }
    //     ack.acknowledge();
    // }
}

import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaConfig {

    // 配置死信队列和错误处理
    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory,
            KafkaTemplate<Object, Object> template) { // 注入KafkaTemplate用于发送死信消息
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);

        // 设置手动提交Offset
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

        // 配置错误处理器：SeekToCurrentErrorHandler 用于重试，达到最大重试次数后交给Recoverer
        // FixedBackOff(interval, maxAttempts) 表示每次重试间隔和最大重试次数
        SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
                new DeadLetterPublishingRecoverer(template), new FixedBackOff(1000L, 2L)); // 失败后重试2次，每次间隔1秒
        factory.setErrorHandler(errorHandler);

        return factory;
    }
}