一、使用java代码生产、消费消息
1、生产者
package com.allwe.client.simple;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
/**
* kafka生产者配置
*
* @Author: AllWe
* @Date: 2024/09/24/17:57
*/
@Slf4j
public class HelloKafkaProducer {
public static void main(String[] args) {
// 设置属性
Properties properties = new Properties();
// 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
// 设置key和value的序列化器，使java对象转换成二进制数组
properties.put("key.serializer", StringSerializer.class);
properties.put("value.serializer", StringSerializer.class);
// new一个生产者producer
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
try {
ProducerRecord<String, String> producerRecord;
try {
// 构建消息
producerRecord = new ProducerRecord<>("topic_1", "student", "allwe");
// 发送消息
producer.send(producerRecord);
System.out.println("消息发送成功");
} catch (Exception e) {
e.printStackTrace();
}
} finally {
// 释放连接
producer.close();
}
}
}
2、消费者
package com.allwe.client.simple;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
* kafka生产者配置
*
* @Author: AllWe
* @Date: 2024/09/24/17:57
*/
@Slf4j
public class HelloKafkaConsumer {
public static void main(String[] args) {
// 设置属性
Properties properties = new Properties();
// 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
// 设置key和value的序列化器，使java对象转换成二进制数组
properties.put("key.deserializer", StringDeserializer.class);
properties.put("value.deserializer", StringDeserializer.class);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
// new一个消费者consumer
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
try {
// 订阅哪些主题，可以多个，推荐订阅一个主题
consumer.subscribe(Collections.singleton("topic_1"));
// 死循环里面实现监听
while (true) {
// 每间隔1s，取一次消息，可能取到多条消息
// 设置一秒的超时时间
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> record : records) {
System.out.println("key:" + record.key() + ",value:" + record.value());
}
}
} finally {
// 释放连接
consumer.close();
}
}
}
3、踩坑
如果连接的不是本机的kafka，需要在目标机器的kafka配置文件中配置真实的ip地址，如果使用默认的配置或者配置为localhost:9092，kafka.clients会将目标机器的ip解析为127.0.0.1，导致连接不上kafka。
二、生产者
1、序列化器
在上面的demo中，由于消息的key和value都是String类型的，就可以使用kafka.client提供的String序列化器，如果想要发送其他自定义类型的对象，可以手动编写一个序列化器和反序列化器，实现Serializer接口，将对象和byte数组互相转换即可。
需要注意的是，生产者使用的自定义序列化器必须和消费者使用的反序列化器对应，否则无法正确解析消息。
那么什么情况下需要使用自定义序列化器呢？
-- 需要兼容一些其他协议。
2、分区器
发送的消息被分配到哪个分区中？分区是如何选择的？假设上面的demo中，主题topic_1有4个分区，分别发送4次消息，处理分区的逻辑是怎样的？
这里需要先配置kafka在创建新的主题时，默认的分区数量，我这里配置为了4。
1）指定分区器
可以选择在创建生产者时，给生产者配置相关的分区器，指定具体分区算法。kafka.client提供了一些分区器，或者自己实现一个分区器。
// 设置分区规则
Properties properties = new Properties();
// 1、默认分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DefaultPartitioner.class);
// 2、统一粘性分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, UniformStickyPartitioner.class);
// 3、自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);自定义分区器：
package com.allwe.client.partitioner;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;
import java.util.List;
import java.util.Map;
/**
* 自定义分区器 - 以value值分区
*/
public class MyPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
List<PartitionInfo> partitionInfoList = cluster.partitionsForTopic(topic);
// 以value值的byte数组处理后再和分区数取模，决定放在哪个分区上
return Utils.toPositive(Utils.murmur2(valueBytes)) % partitionInfoList.size();
}
@Override
public void close() {
}
@Override
public void configure(Map<String, ?> map) {
}
}
2）指定分区
也可以选择在构建消息时指定分区，此时的分区优先级最高，不会被其他分区器影响。
# 创建消息时指定分区为 0
ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic_1", 0, "student", "allwe");3、生产者发送消息的回调
package com.allwe.client.partitioner;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.Future;
/**
* kafka生产者配置 - 自定义分区器 & 发送消息回调
*
* @Author: AllWe
* @Date: 2024/09/24/17:57
*/
@Slf4j
public class PartitionerProducer {
public static void main(String[] args) {
// 设置属性
Properties properties = new Properties();
// 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
// 设置key和value的序列化器，使java对象转换成二进制数组
properties.put("key.serializer", StringSerializer.class);
properties.put("value.serializer", StringSerializer.class);
// 设置自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
// new一个生产者producer
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
try {
ProducerRecord<String, String> producerRecord;
try {
// 构建指定分区的消息，此时指定的分区不会变
// producerRecord = new ProducerRecord<>("topic_1", 0, "student", "allwe");
for (int i = 0; i < 10; i++) {
// 构建消息
producerRecord = new ProducerRecord<>("topic_2", "student", "allwe" + i);
// 发送消息
Future<RecordMetadata> future = producer.send(producerRecord);
// 解析回调元数据
RecordMetadata recordMetadata = future.get();
System.out.println(i + ",offset:" + recordMetadata.offset() + ",partition:" + recordMetadata.partition());
}
} catch (Exception e) {
e.printStackTrace();
}
} finally {
// 释放连接
producer.close();
}
}
}
打印结果：
4、异步解析生产者发送消息的回调
package com.allwe.client.callBack;
import com.allwe.client.partitioner.MyPartitioner;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
/**
* kafka生产者配置 - 异步解析发送消息回调
*
* @Author: AllWe
* @Date: 2024/09/24/17:57
*/
@Slf4j
public class AsynPartitionerProducer {
public static void main(String[] args) {
// 设置属性
Properties properties = new Properties();
// 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
// 设置key和value的序列化器，使java对象转换成二进制数组
properties.put("key.serializer", StringSerializer.class);
properties.put("value.serializer", StringSerializer.class);
// 设置自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
// new一个生产者producer
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
try {
ProducerRecord<String, String> producerRecord;
try {
for (int i = 0; i < 10; i++) {
// 构建消息
producerRecord = new ProducerRecord<>("topic_3", "student", "allwe" + i);
// 发送消息, 设置异步回调解析器
producer.send(producerRecord, new CallBackImpl());
}
System.out.println("发送完成,topic_4");
} catch (Exception e) {
e.printStackTrace();
}
} finally {
// 释放连接
producer.close();
}
}
}
package com.allwe.client.callBack;
import cn.hutool.core.util.ObjectUtil;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
* 异步发送消息回调解析器
*/
public class CallBackImpl implements Callback {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (ObjectUtil.isNull(e)) {
// 解析回调元数据
System.out.println("offset:" + recordMetadata.offset() + ",partition:" + recordMetadata.partition());
} else {
e.printStackTrace();
}
}
}
5、生产者缓冲
1）为什么kafka在客户端发送消息的时候需要做一个缓冲？
① 减少IO的开销（单个 -> 批次），需要修改配置文件。
② 减少GC（核心）。
2）如何配置缓冲？
producer.properties配置文件中修改下面两个参数：
消息的大小：batch.size = 默认16384(16K)
暂存的时间：linger.ms = 默认0ms
上面两个条件只要达到一个，就会发送消息，所以在默认配置下，生产一条消息就立即发送。
3）减少GC的原理
producer.properties配置文件的参数：
缓冲池大小：buffer.memory = 默认32M
kafka客户端使用了缓冲池，默认大小32M，当有一条新的消息进入缓冲池，达到了任何一个条件后就发送。发送后不用立即回收内存，而是初始化一下缓冲池即可，减少了GC的次数。
简单说就是利用池化技术减少了对象的创建 -> 减少内存分配次数 -> 减少了垃圾回收次数。
4）使用缓冲池的风险
当缓存的消息超出缓冲池的大小，kafka就会抛出OOM异常。
如果写入消息太快，但是上一次send方法没有执行完，就会导致上一次缓存的消息不能删除，这一次进来的消息又太多，最终写满了缓冲池，触发OOM异常。
解决办法就是适当调整buffer.memory参数和batch.size参数，增加缓冲池大小，缩小每一批次的大小。
三、Kafka Broker
消息从生产者发送出去后，就进入了broker中。在kafka broker中，每一个分区就是一个文件。
四、消费者
1、消费者群组
在消费的过程中，一般情况下使用群组消费，设置group_id_config。
核心：kafka群组消费的负载均衡建立在分区级别。
1）单个群组场景
一个分区只能由一个消费者消费。
在kafka执行过程中，支持动态添加或者减少消费者。
2）多个群组场景
群组之间的消费是互不干扰的，比如群组A的消费者和群组B的消费者可以同时消费同一个分区的消息。
2、Demo记录
写一个生产者，我为了测试顺畅写了一个无限循环的。只启动一次，输入参数即可实现批量发送消息。
package com.allwe.client.singleGroup;
import com.allwe.client.partitioner.MyPartitioner;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.Scanner;
/**
* kafka生产者配置 - 无限生产消息
*
* @Author: AllWe
* @Date: 2024/09/24/17:57
*/
@Slf4j
public class Producer {
public static void main(String[] args) {
// 设置属性
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
properties.put("key.serializer", StringSerializer.class);
properties.put("value.serializer", StringSerializer.class);
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
// new一个生产者producer
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
Scanner scanner = new Scanner(System.in);;
try {
int count;
while (true) {
System.out.println("==================输入消息条数===================");
String nextLine = scanner.nextLine();
if ("exit".equals(nextLine)) {
break;
}
count = Integer.parseInt(nextLine);
ProducerRecord<String, String> producerRecord;
try {
for (int i = 0; i < count; i++) {
// 构建消息
producerRecord = new ProducerRecord<>("topic_5", "topic_5", "allwe" + i);
producer.send(producerRecord);
}
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("发送完成,topic_5");
}
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
// 释放连接
producer.close();
scanner.close();
}
}
}
写一个消费者base类，由于测试消费者需要启动很多类，我这里为了方便写了一个baseConsumer类，调用时new这个类的对象即可调用消费方法。
package com.allwe.client.singleGroup;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
* kafka 消费者配置
*
* @Author: AllWe
* @Date: 2024/09/24/17:57
*/
@Slf4j
@Data
public class SingleGroupBaseConsumer {
private String groupIdConfig;
private String topicName;
private KafkaConsumer<String, String> consumer;
public SingleGroupBaseConsumer(String groupIdConfig, String topicName) {
this.groupIdConfig = groupIdConfig;
this.topicName = topicName;
createConsumer();
}
private void createConsumer() {
// 设置属性
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
properties.put("key.deserializer", StringDeserializer.class);
properties.put("value.deserializer", StringDeserializer.class);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);
consumer = new KafkaConsumer<>(properties);
}
public void poll() {
try {
consumer.subscribe(Collections.singleton(topicName));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
int count = 0;
for (ConsumerRecord<String, String> record : records) {
count = 1;
System.out.println("partition:" + record.partition() + ",key:" + record.key() + ",value:" + record.value());
}
if (count == 1) {
// 消费到消息了就打印分隔线
System.out.println("===============================");
}
}
} finally {
consumer.close();
}
}
}
1）单个群组场景
群组id：allwe01
package com.allwe.client.singleGroup;
import lombok.extern.slf4j.Slf4j;
/**
* kafka消费者启动器
*
* @Author: AllWe
* @Date: 2024/09/24/17:57
*/
@Slf4j
public class SingleGroupConsumer_1 {
public static void main(String[] args) {
SingleGroupBaseConsumer singleGroupBaseConsumer = new SingleGroupBaseConsumer("allwe01", "topic_5");
singleGroupBaseConsumer.poll();
}
}
我这里只放了一个消费者的消费记录，根据消费者控制台打印的数据，可以看到两条信息：
① 该消费者只能消费分区=1的消息。
② 消费者消费消息时，每次拿到的消息数量不确定。
2）多个群组场景
群组id：allwe02
package com.allwe.client.group;
import com.allwe.client.singleGroup.SingleGroupBaseConsumer;
import lombok.extern.slf4j.Slf4j;
/**
* kafka消费者启动器
*
* @Author: AllWe
* @Date: 2024/09/24/17:57
*/
@Slf4j
public class GroupConsumer_1 {
public static void main(String[] args) {
SingleGroupBaseConsumer singleGroupBaseConsumer = new SingleGroupBaseConsumer("allwe02", "topic_5");
singleGroupBaseConsumer.poll();
}
}
可以看到，这里新加入了一个消费者群组，只有一个消费者，它就消费到了全部分区的消息。
3、ACK确认
消费者在成功消费消息后，会进行ACK确认。提交最后一次消费消息的偏移量，下一次消费就从上次提交的偏移量开始，如果一个新的消费者群组消费一个主题的消息，可以根据不同的配置来指定起始的偏移量。
// 从最早的消息开始消费
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// 从已提交的偏移量开始消费 - 默认配置
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");在kafka内部，有一个名字叫【__consumer_offsets】的主题，保存了消费者对各个主题的消费偏移量。消费者每一次发送的ACK确认，都会更新这个主题中的偏移量数据。
1）自动提交ACK的消费模式
默认的消费模式。
只要拿到了消息，就自动提交ACK确认。
但是有一个风险，就是虽然消费者成功取到了消息，但是在程序处理过程中出现了异常，同时提交了ACK确认，那么这条消息就永远不会被正确地处理。
所以有时候我们需要避免自动提交ACK确认，改成手动提交ACK确认。
2）手动提交ACK确认
取消自动提交
// 取消自动提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);① 同步提交
// 同步提交ACK确认 - 提交不成功就一直重试，成功后才会继续往下执行
consumer.commitSync();立刻进行ACK确认。但是容易造成阻塞，只有等待ACK确认成功后，才会继续执行程序。如果ACK确认不成功，就会一直重试。
② 异步提交
// 异步提交ACK确认
consumer.commitAsync();异步提交不会阻塞应用程序，提交失败不会重试提交。
③ 组合使用demo
public void poll() {
try {
consumer.subscribe(Collections.singleton(topicName));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
int count = 0;
for (ConsumerRecord<String, String> record : records) {
count = 1;
System.out.println("partition:" + record.partition() + ",offset:" + record.offset() +",key:" + record.key() + ",value:" + record.value());
}
if (count == 1) {
// 消费到消息了就打印分隔线
System.out.println("===============================");
}
// 异步提交ACK确认
consumer.commitAsync();
}
} finally {
try {
// 同步提交ACK确认 - 提交不成功就一直重试，成功后才会继续往下执行
consumer.commitSync();
} finally {
consumer.close();
}
}
}3）手动批量提交ACK确认
如果消费者在某一时刻取到的消息数量太多，那么给每一条消息单独提交ACK确认太浪费资源，可以选择批量提交ACK确认。核心思想就是在程序中暂存偏移量，达到设定的阈值后就触发批量提交。
kafka.Consumer提供的异步提交ACK方法支持批量提交。
