ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀，每一条都经过真实交付验证。

首页 / 资讯中心 / 正文

Kafka核心参数(带完善)

2026/9/12 12:24:38 拓冰建站 浏览量
Kafka核心参数(带完善)

客户端

api

Kafka提供了以下两套客户端API

  • HighLevel(重点)
  • LowLevel 

HighLevel API封装了kafka的运行细节，使用起来比较简单，是企业开发过程中最常用的客户端API。 而LowLevel API则需要客户端自己管理Kafka的运行细节，Partition，Offset这些数据都由客户端自行管理。这层API功能更灵活，但是使用起来非常复杂，也更容易出错。只在极少数对性能要求非常极致的场景才会偶尔使用

生产者发送消息

发送流程:

  1. 组装生产者核心配置参数
  2. 初始化生产者
  3. 组装消息
  4. 发送消息, 三种模式
    1. 单向发送, 不等待broker返回结果
    2. 同步发送
    3. 异步发送
  5. 关闭生产者

代码:

package com.kk.kafka.demo;import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;public class ProducerTest {public static final String KAFKA_URL = "192.168.6.128:9092";public static final String TOPIC = "oneTopic";public static void main(String[] args) throws ExecutionException, InterruptedException {// 组装生产者配置Properties ps = new Properties();ps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_URL);ps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");ps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");// 初始化生产者Producer<String, String> producer = new KafkaProducer<>(ps);for (int i = 0; i < 5; i++) {ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC, "key" + i, "message" + i);// 同步发送producer.send(producerRecord);// 同步发送RecordMetadata metadata = producer.send(producerRecord).get();//异步发送producer.send(producerRecord, new Callback() {@Overridepublic void onCompletion(RecordMetadata metadata, Exception e) {if (metadata != null) {System.out.println("Message sent successfully! Topic: " + metadata.topic() +", Partition: " + metadata.partition() +", Offset: " + metadata.offset() +", message: " + producerRecord.value());} else {System.err.println("Error sending message: " + e.getMessage());}}});}producer.close();}
}

消费者消费消息

消费流程:

  1. 组装消费者核心配置参数
  2. 初始化消费者
  3. 订阅topic, 可订阅多个
  4. 拉取消息, 可配置超时时间
  5. 提交offset, 分为同步和异步两种方式, 服务端维护offset消费进度

代码:

package com.kk.kafka.demo;import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;public class ConsumerTest {public static final String KAFKA_URL = "192.168.6.128:9092";public static final String TOPIC = "oneTopic";public static void main(String[] args) {// 组装消费者配置参数Properties props = new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_URL);props.put(ConsumerConfig.GROUP_ID_CONFIG, "your-consumer-group");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");// 初始化消费者Consumer<String, String> consumer = new KafkaConsumer<>(props);// 订阅topicconsumer.subscribe(Arrays.asList(TOPIC));while (true) {// 拉取消息,  100毫秒超时时间ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(100));//处理消息for (ConsumerRecord<String, String> record : records) {System.out.println("start Consumer offset = " + record.offset() + ";key = " + record.key() + "; value= " + record.value());}//提交offset，消息就不会重复推送。//同步提交，表示必须等到offset提交完毕，再去消费下一批数据。consumer.commitSync();//异步提交，表示发送完提交offset请求后，就开始消费下一批数据了。不用等到Broker的确认。
//            consumer.commitAsync();}}
}

客户端整体流程

拦截器

序列化器

发送到Dequeue

Dequeue满了或者批次满了或者阈值时间 推到InflightRequest

send线程将InflightRequest推到服务端Partition, 满足一定阈值

缓存机制

broker给生产者ack

消费者分组策略

相关消费者配置

  • GROUP_ID_CONFIG: 群组唯一id
  • GROUP_INSTANCE_ID_CONFIG: 消费实例id, 可以减少不必要rebalance

生产者给Topic投递消息, 消息会均匀的存到partition. 消息会向所有订阅该Topic的消费实例推送, 推送时, 一个消费群组只会推送一份. 也就是, 同一个群组里面, 只会有一个消费实例能消费; 不同消费群组可以重复消费消息.Offset就是记录每个消费群组在partition的处理消息进度

offset丢失怎么解决

1. offset丢失:

  1. 初始化consumer group时, 设置offset失败
  2. offset对应的数据文件被删除

解决: 服务端有个兜底方案, 可以配置消费者配置

ConsumerConfig.AUTO_OFFSET_RESEWT_CONFIG ：当Server端没有对应的Offset时，要如何处理。

可选项：

  • earliest： 自动设置为当前最早的offset
  • latest：自动设置为当前最晚的offset
  • none： 如果消费者组对应的offset找不到，就向Consumer抛异常。
  • 其他选项： 向Consumer抛异常。

2. offset不一致问题

消费者可以选择同步提交或者异步提交

  • 同步提交: 消息处理完, 提交. 消息处理失败, 选择不提交, 等重试. 如果消费过慢, 服务端不会无限等, 会认为本次消费失败, 会给同组的其他消费实例投递消费, 可能存在重复消费问题 
  • 异步提交: 先提交, 后处理消息. 如果消息处理失败, offset又被提交, 就存在客户端与服务端offset不一致问题

解决: 可以交由客户端管理offset, 存到redis或者mysql等中间件.客户端可以控制消息处理进度, 实时推进offset

生产者拦截器

生产者配置: INTERCEPTOR_CLASSES_CONFIG

        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.kk.kafka.demo.MyProducerInterceptor");


public class MyProducerInterceptor implements ProducerInterceptor {// 发送消息触发@Overridepublic ProducerRecord onSend(ProducerRecord producerRecord) {System.out.println("onSend producerRecord:" + producerRecord.toString());return producerRecord;}// 收到服务端相应触发@Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {System.out.println("acknowledgement recordMetadata:" + recordMetadata.toString());}// 生产者连接关闭触发@Overridepublic void close() {System.out.println("producer close");}// 整理配置项@Overridepublic void configure(Map<String, ?> map) {}
}

消息序列化

相关配置:

  • KEY_SERIALIZER_CLASS_CONFIG: key序列号
  • VALUE_SERIALIZER_CLASS_CONFIG: 消息value序列号
  • KEY_DESERIALIZER_CLASS_CONFIG: key反序列化
  • VALUE_DESERIALIZER_CLASS_CONFIG: 消息value反序列化

1. 生产者:

ps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
ps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

2. 消费者

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

消息分区路由机制

生产者缓存机制

其他重要核心参数 

幂等性原理

事务消息