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.设置消费
@ComponentpublicclassKafkaConsumer{privatefinalstaticString TOPIC_NAME="topic_NAME";@KafkaListener(topics = TOPIC_NAME,groupId ="defaultName")publicvoidlistenGroup(ConsumerRecord<String,String>record,Acknowledgment ack){System.out.println(record.value());System.out.println(record);手动提交offset
ack.acknowledge();}}
本文转载自: https://blog.csdn.net/ai_haibin/article/details/131087448
版权归原作者 听者listener 所有, 如有侵权,请联系我们删除。
版权归原作者 听者listener 所有, 如有侵权,请联系我们删除。