springboot简单使用kafka消费者监听，以及kafka配置账号密码-Toy模板网

这篇具有很好参考价值的文章主要介绍了springboot简单使用kafka消费者监听，以及kafka配置账号密码。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1.引入依赖

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

2.yml配置

 spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    properties:
      security:
        protocol: SASL_PLAINTEXT
      sasl:
        mechanism: SCRAM-SHA-512
        jaas:
          config: org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";
 
      #producer:
      #当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
      #如果没收到ack响应 重试次数 设置大于0的值，则客户端会将发送失败的记录重新发送
    #retries: 3
      # 每次批量发送消息的数量,produce积累到一定数据，一次发送 每次提交的批次大小 16K
      # batch-size: 16384
      #produce积累数据一次发送，缓存大小达到buffer.memory就发送数据 32M
      # buffer-memory: 33554432
      # 0 是直接响应返回  1是leader完成响应返回 -1(all) 是ISR里 leade follower 全部完成 响应
      #procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
      #acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
      #acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
      #acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
      #可以设置的值为：all, -1, 0, 1
      # acks: 1
      #ey value 的序列化
      # key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # value-serializer: org.apache.kafka.common.serialization.StringSerializer

    consumer:
      # 指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名
      group-id: defaultName
      #关闭自动提交
      enable-auto-commit: false
      #重置消费者的offset
      # smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
      auto-offset-reset: latest
      #key value 的反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      max-poll-records: 5
    listener:
      # RECORD 当每一条记录被消费者监听器（ListenerConsumer）处理后提交
      # BATCH 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理后提交
      # TIME 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理后,距离上次提交时间大于TIME时提交
      # COUNT 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理后,被处理record数量大于COUNT时提交
      # COUNT_TIME TIME | COUMT 有一个条件满足时提交
      # MANUAL 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理后,手动调用 Acknowledgment.acknowledge()后提交
      # MANUAL_IMMEDIATE 手动调用 Acknowledgment.acknowledge() 之后 立即提交
      ack-mode: manual_immediate
      # 消费监听接口监听的主题不存在时，默认会报错
      missing-topics-fatal: false

3.设置消费

@Component
public class KafkaConsumer {

    private final static  String TOPIC_NAME="topic_NAME";

    @KafkaListener(topics = TOPIC_NAME,groupId = "defaultName")
    public void listenGroup(ConsumerRecord<String,String> record, Acknowledgment ack){
        System.out.println(record.value());
        System.out.println(record);
        
        手动提交offset
        ack.acknowledge();
      }
    }