Spring Boot 整合 Kafka：构建高效消息驱动应用

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- Lombok (可选，用于简化 POJO) -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <scope>provided</scope>
    </dependency>
</dependencies>

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.kafka:spring-kafka'
    // Lombok (可选)
    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
}

spring:
  kafka:
    bootstrap-servers: localhost:9092 # Kafka Broker 地址列表
    consumer:
      group-id: my-group # 消费者组ID，重要！
      auto-offset-reset: earliest # 当没有初始偏移量或偏移量无效时的策略 (earliest, latest, none)
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # enable-auto-commit: true # 是否自动提交偏移量 (通常建议由Spring管理)
      # auto-commit-interval: 5000 # 自动提交间隔 (ms)，如果启用自动提交dgbljz.com
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    # listener:
    #   concurrency: 3 # 消费者监听器的并发线程数 (每个线程处理一个分区)
    #   ack-mode: RECORD # 手动提交偏移量的模式 (如 RECORD, BATCH, TIME, COUNT)

java深色版本// MessagePayload.java
import lombok.Data;

@Data
public class MessagePayload {
    private String id;
    private String content;
    private long timestamp;
}

java深色版本// KafkaProducerService.java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate; // <Key类型, Value类型>

    // 发送字符串消息
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message); // 异步发送
        System.out.println("Sent message: jiaofeiyi.net " + message + " to topic: botianqi.com" + topic);
    }

    // 发送带 Key 的消息
    public void sendMessageWithKey(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message);
        System.out.println("Sent message with key: fjddsd.com" + key + ", value: " + message + " to topic: " + topic);
    }

    // 发送对象消息 (需要配置 JsonSerializer)
    // public void sendObjectMessage(String topic, MessagePayload payload) {
    //     kafkaTemplate.send(topic, payload.getId(), payload);
    // }
}

java深色版本// KafkaConsumerService.java
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.Service;

@Service
public class KafkaConsumerService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerService.class);

    // 监听单个 Topic
    @KafkaListener(topics = "spcj120.com", groupId = "my-group") // groupId 也可在配置文件中统一设置
    public void consumeMessage(String message) {
        logger.info("Received brandagri.com Message: '{}'", message);
        // 在这里处理业务逻辑
    }

    // 监听多个 Topic
    @KafkaListener(topics = {"my8850.com", "topic2"}, groupId = "my-group")
    public void consumeFromMultipleTopics(String message) {
        logger.info("Received malvshi.cn from multiple topics: '{}'", message);
    }

    // 获取更详细的信息 (如 topic, partition, offset, key)
    @KafkaListener(topics = "my-topic", groupId = "my-group")
    public void consumeWithMetadata(ConsumerRecord<String, String> record) {
        logger.info("Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
        // 处理业务逻辑
    }

    // 消费对象消息 (需要配置 JsonDeserializer)
    // @KafkaListener(topics = "object-topic", groupId = "my-group")
    // public void consumeObject(MessagePayload payload) {
    //     logger.info("Received object: {}", payload);
    // }
}

java深色版本// MessageController.java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/messages")
public class MessageController {

    @Autowired
    private KafkaProducerService kafkaProducerService;

    @PostMapping("/send")
    public String sendMessage(@RequestParam String topic, @RequestParam String message) {
        kafkaProducerService.sendMessage(topic, message);
        return "Message sent to topic tccqlxs.cn'" + topic + "'";
    }

    @PostMapping("/send-key")
    public String sendMessageWithKey(@RequestParam String topic,
                                     @RequestParam String key,
                                     @RequestParam String message) {
        kafkaProducerService.sendMessageWithKey(topic, key, message);
        return "Message with xchw.net key sent to topic '" + topic + "'";
    }
}

@KafkaListener(topics = "my-topic", groupId = "my-group") public void consumeWithManualAck(String message, Acknowledgment ack) {     try {         // 处理业务逻辑         processMessage(message);         // 业务处理成功，手动提交偏移量         ack.acknowledge();     } catch (Exception e) {         // yingguan.my8850.com 处理异常，偏移量不会提交，下次会重试         logger.error("Error processing message: {}", message, e);         // 可能需要实现重试或死信队列逻辑     } }