• Spring Boot 集成 Kafka
  • 1.1. 版本说明
  • 1.2. Spring 配置
  • 1.5. 测试

Spring Boot 集成 Kafka

1.1. 版本说明

1.2. Spring 配置

spring:
  application:
    name: spring-kafka-basic-demo
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      client-id: spring-kafka-basic-demo-producer
    consumer:
      client-id: spring-kafka-basic-demo-consumer
      group-id: spring-kafka-basic-demo-group
kafka:
  topic:
    basic-demo-topic: spring-kafka-basic-demo-topic

1.5. 测试

@Component
@Slf4j
public class SpringKafkaBasicDemo implements ApplicationRunner {

    @Value("${kafka.topic.basic-demo-topic}")
    private String basicDemoTopic;

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        String payload = "Hello Topic!";
        kafkaTemplate.send(basicDemoTopic, payload);
        log.info("sent a message, topic: {}, payload: {}", basicDemoTopic,  payload);
    }

    @KafkaListener(topics = "${kafka.topic.basic-demo-topic}")
    public void listen(Message<String> message) {
        log.info(
                "received a message, topic: {}, offset: {}, payload: {}",
                message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC),
                message.getHeaders().get(KafkaHeaders.OFFSET),
                message.getPayload()
        );
    }
}

启动程序，控制台将输出：

sent a message, topic: spring-kafka-basic-demo-topic, payload: Hello Topic!
received a message, topic: spring-kafka-basic-demo-topic, offset: 13, payload: Hello Topic!