Kafka使⽤规范
Kafka
区别
配置
>>#【Kafka集】>>#
spring.kafka.bootstrap-servers: 172.16.74.46:9092,172.16.171.152:9092,172.16.72.30:9092,172.16.73.205:9092,172.16.75.18:9092 >>#【初始化⽣产者配置】>>#
# 重试次数
spring.ies: 0
# 应答级别：多少个分区副本备份完成时向⽣产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks: 1
# 批量⼤⼩
spring.kafka.producer.batch-size: 16384
# 提交延时
spring.kafka.producer.properties.linger.ms: 0
# 当⽣产端积累的消息达到batch-size或接收到消息linger.ms后,⽣产者就会将消息提交给kafka
# linger.ms为0表⽰每接收到⼀条消息就提交给kafka，这时候batch-size其实就没⽤了
# ⽣产端缓冲区⼤⼩
spring.kafka.producer.buffer-memory :  33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer: org.apache.kafkamon.serialization.StringSerializer
spring.kafka.producer.value-serializer: org.apache.kafkamon.serialization.StringSerializer
# ⾃定义分区器
# spring.kafka.producer.properties.partitioner.class: com.felix.kafka.producer.CustomizePartitioner
>>#【初始化消费者配置】>>#
# 默认的消费组ID
up.id: defaultConsumerGroup
# 是否⾃动提交offset
able-auto-commit: false
# 提交offset延时(接收到消息后多久提交offset)
sumer.automit.interval.ms: 1000
# 当kafka中没有初始offset或offset超出范围时将⾃动重置offset
# earliest：重置为分区中最⼩的offset;
springboot其实就是spring# latest：重置为分区中最新的offset(消费分区中新产⽣的数据);
# none：只要有⼀个分区不存在已提交的offset，就抛出异常;
sumer.auto-offset-reset: latest
# 消费会话超时时间(超过这个时间consumer没有发送⼼跳，就会触发rebalance操作)
sumer.properties.session.timeout.ms: 120000
# 消费请求超时时间
quest.timeout.ms: 180000
# Kafka提供的序列化和反序列化类
sumer.key-deserializer: org.apache.kafkamon.serialization.StringDeserializer
sumer.value-deserializer: org.apache.kafkamon.serialization.StringDeserializer
# 消费端监听的topic不存在时，项⽬启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal: false
# 设置批量消费
# spring.pe: batch
# 批量消费每次最多消费多少条消息
# sumer.max-poll-records: 50
task.pic: rm-bp160q69ui8c010b1
# group: ${task.up}-${spring.profiles.active}
task.up: soul-task-platform-feedback-group
Kafka中bootstrap-server、broker-list和zookeeper的区别
*，
-
-bootstrap-server
--zookeeper
Spring Boot 整合——kafka消费模式AckMode以及⼿动消费
*，
MANUAL
MANUAL_IMMEDIATE
RECORD
BATCH
TIME
COUNT
COUNT_TIME
/
**
* MANUAL  当每⼀批poll()的数据被消费者（ListenerConsumer）处理之后, ⼿动调⽤Acknowledgment.acknowledge()后提交    * @param consumerFactory
* @return
*/
@Bean("manualListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> manualListenerContainerFactory(
ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
//
factory.setBatchListener(true);
// 消息过滤策略
factory.setRecordFilterStrategy(consumerRecord -> {
if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {
return false;
}
//返回true消息则被过滤
return true;
});
/
/配置⼿动提交offset
return factory;
}
/**
* MANUAL  当每⼀批poll()的数据被消费者（ListenerConsumer）处理之后, ⼿动调⽤Acknowledgment.acknowledge()后提交    * @param message
* @param ack
*/
@KafkaListener(containerFactory = "manualListenerContainerFactory" , topics = "kafka-manual")
public void onMessageManual(List<Object> message, Acknowledgment ack){
log.info("manualListenerContainerFactory 处理数据量：{}",message.size());
message.forEach(item -> log.info("manualListenerContainerFactory 处理数据内容：{}",item));
ack.acknowledge();//直接提交offset
}

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。

IDEA端口占用,启动失败,提示Webserverfailedtostart.Port80。。。
« 上一篇
java获取linux服务器上的IP操作
下一篇 »

发表评论