目录

目标

快速整合

配置kafka核心参数

生产者发送消息的不同方式

1. 根据SpringBoot官方文档整合kafka，实现生产者与消费者的交互；
2. 掌握核心kafka核心参数的配置。

快速整合

第一步：创建一个SpringBoot项目，我用的SpringBoot版本是2.7.0（略）。

第二步：kafka的相关依赖。

  org.springframework.kafka
  spring-kafka

第三步：在application.yml文件中配置kafka服务器的地址。

spring:
  kafka:
    #多个节点用逗号隔开
    bootstrap-servers: localhost:9092

第四步：指定主题发送一个消息。

package com.ctx.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;

@RestController
public class MyProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    //iphoneTopic主题
    private static final String IPHONE_TOPIC="iphoneTopic";

    
    @PostMapping("/simpleSend")
    public void simpleSend(@RequestBody Map msg){
        //send(主题，发送的内容)
        kafkaTemplate.send(IPHONE_TOPIC, msg.toString());
    }
}

第五步：配置kafka监听器来消费消息。

package com.ctx.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MyConsumer {
    //iphoneTopic主题
    private static final String IPHONE_TOPIC="iphoneTopic";
    //消费者组
    private static final String APPLE_GROUP="apple_group";

    
    @KafkaListener(topics = IPHONE_TOPIC,groupId = APPLE_GROUP)
    public void simpleSend(ConsumerRecord record){
        System.out.println("内容："+record.value());
        System.out.println("分区："+record.partition());
        System.out.println("偏移量："+record.offset());
        System.out.println("创建消息的时间戳："+record.timestamp());
    }
}

第六步：调用发送消息的接口，观察控制台的输出情况。

---

配置kafka核心参数

server:
  port: 8099
spring:
  kafka:
    #指定Broker，多个Broker组成的集群用逗号分开。
    bootstrap-servers: localhost:9092
    #生产者
    producer:
      #acks=all或者acks=-1
      #此时要看min.insync.replicas（最小同步副本数量，默认等于1）参数设定的值。例如：min.insync.replicas=2，表示需要有2个kafka节点写入数据才会返回客户端提示成功。推荐大于等于2。
      #acks=0
      #生产者不需要等待kafka节点回复确认收到消息，就可以继续发送下一条消息。
      #acks=1
      #消息写入Leader（主节点），但是不需要等待其他Follower（从节点）同步消息，就可以继续发送下一条消息。如果Follower没有同步到消息，而Leader宕机，则消息丢失。这种模式等同于acks=all或者acks=-1，且min.insync.replicas=1。
      acks: 1
      #生产者向消费者发送消息失败后允许重试的次数，重试次数范围：[0,2147483647]； 如果要开启重试，需要将enable.idempotence设置为false或1。
      retries: 3
      #批量发送消息的大小，单位字节，1024Byte * 16= 16KB
      batch-size: 16384
      #发送消息的缓冲区大小，默认33554432，即32MB。
      buffer-memory: 33554432
      #key-value的序列化器，发送消息时，消息在网络中传输，所以需要把消息序列化。
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      #假如批量发送消息设置的批次（batch-size）为16KB，如果经过了10毫秒发送方仍然没有获取到16KB批次的消息，则直接发送消息。
      properties:
        linger.ms: 10
    #消费者
    consumer:
      #key-value的序列化器，发送消息时，消息在网络中传输，所以需要把消息序列化。
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      #是否自动提交偏移量
      enable-auto-commit: false
      #如果开启了自动提交，则该配置生效，其值表示消费者自动提交的频率，单位毫秒。默认事件是5秒。
      #auto-commit-interval: 5000
      #是从最早的消息开始消费，还是实时消费。earliest：从最早的消费开始消费。latest：从最新的消息开始消费。
      auto-offset-reset: latest
      properties:
        #单位为毫秒，默认45秒，即Broker在设定时间内感知不到该消费者，则消费者组重新平衡，消费者被消费者组剔除，分区重新分配给其他消费者。这个值的设定范围在group.min.session.timeout.ms和group.max.session.timeout.ms之间，即消费者的最大最小会话时间，这两个默认值分别是6秒和30分钟。
        session.timeout.ms: 45000
        #消费者处理消息超时时间，即消费者拉取到消息后，如果超过该限定时间则消费者组将重新平衡，把分区分配给其他消费者。默认值是5分钟。
        max.poll.interval.ms: 300000

---

生产者发送消息的不同方式

package com.ctx.producer;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;


@RestController
public class MyProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    //iphoneTopic主题
    private static final String IPHONE_TOPIC="iphoneTopic";

    
    @PostMapping("/simpleSend")
    public void simpleSend(@RequestBody Map msg){
        //send(主题,发送的内容)
        kafkaTemplate.send(IPHONE_TOPIC, msg.toString());
    }

    
    @PostMapping("/setKeySend")
    public void setKeySend(@RequestBody Map msg){
        //send(主题,key,发送的内容)
        kafkaTemplate.send(IPHONE_TOPIC, msg.get("id"),msg.toString());
    }

    
    @PostMapping("/setTimeSend")
    public void setTimeSend(@RequestBody Map msg){
        //send(主题,分区,时间戳,key,发送的内容)
        kafkaTemplate.send(IPHONE_TOPIC, 0,1653837059222L,msg.get("id"),msg.toString());
    }

    
    @PostMapping("/setProducerRecordSend")
    public void setProducerRecordSend(@RequestBody Map msg){
        //我这里设置主题和发送的消息，具体的构造方法还有很多，这里不赘述。
        ProducerRecord producerRecord = new ProducerRecord<>(
                IPHONE_TOPIC, msg.toString()
        );
        kafkaTemplate.send(producerRecord);
    }

}