阅读 213

SpringBoot整合kafka

此文章已同步更新至我的个人博客simonting.gitee.io

准备工作

  • 安装kafka,版本为2.6.0
  • 安装zookeeper,版本为3.6.2
  • 启动kafka
  • 启动zookeeper
  • SpringBoot版本2.3.1.RELEASE

核心目录结构

  • KafkaConsumer.java 消费者
  • KafkaProducerController 生产者
  • application.yaml 主配置文件
  • application-kafka.properties kafka配置文件

代码

pom依赖

<dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
      <version>2.6.1</version>
</dependency>
复制代码

application-kafka.properties

#kafka地址 brokers集群地址用,隔开
spring.kafka.bootstrap-servers=127.0.0.1:9092

#生产者的配置
spring.kafka.producer.batch-size=16
#发送失败重试次数
spring.kafka.producer.retries=0
#即32MB的批处理缓冲区
spring.kafka.producer.buffer-memory=33554432
#key序列化方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
#value序列化方式
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

#消费者的配置
#Kafka中没有初始偏移或如果当前偏移在服务器上不再存在时,默认区最新 ,有三个选项:latest, earliest, none
spring.kafka.consumer.auto-offset-reset=latest
#是否开启自动提交
spring.kafka.consumer.enable-auto-commit=true
#自动提交的时间间隔
spring.kafka.consumer.auto-commit-interval=100
#key的解码方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#value的解码方式
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#在kafka/config文件的consumer.properties中有配置
spring.kafka.consumer.group-id=test-consumer-group
#批量一次最大拉取数据量
spring.kafka.consumer.max-poll-records=1000

#topic
topicName.topic1=test1
topicName.topic2=test2
复制代码

KafkaProducer

@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaProducerController {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @GetMapping("/send")
    public ResponseEntity<?> send(@RequestParam String topic, @RequestParam String msg) {
        kafkaTemplate.send(topic, msg);
        log.info("[Producer] topic: {}; msg: {}", topic, msg);
        return new ResponseEntity<>(HttpStatus.OK);
    }
}
复制代码

KafkaConsumer

@Slf4j
@Component
public class KafkaConsumer {

    @KafkaListener(topics = "${topicName.topic1}", groupId = "${spring.kafka.consumer.group-id}")
    @KafkaListener(topics = "${topicName.topic2}", groupId = "${spring.kafka.consumer.group-id}")
    public void listen(ConsumerRecord<?, ?> record) {
        log.info("[Consumer] topic: {}; msg: {}; offset: {}", record.topic(), record.value(), record.offset());
    }
}
复制代码

或者下面的这种写法,两种写法都能同时监听到多个topic的消息

@Slf4j
@Component
public class KafkaConsumer {

    @KafkaListeners({
            @KafkaListener(topics = "${topicName.topic1}"),
            @KafkaListener(topics = "${topicName.topic2}")
    })
    public void listen(ConsumerRecord<?, ?> record) {
        log.info("[Consumer] topic: {}; msg: {}; offset: {}", record.topic(), record.value(), record.offset());
    }
}
复制代码

测试

使用postman向test1主题与test2主题分别发送一条消息,查看日志。

日志:

测试成功!

消费者成功的接收到了生产者发送的消息。