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();

    }

}

免责申明：

本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！

《数据治理行业实践白皮书》下载地址：https://fs80.cn/4w2atu

《数栈V6.0产品白皮书》下载地址：https://fs80.cn/cw0iw1

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack