一、使用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方法支持批量提交。
五、多线程安全问题
1、多线程安全的定义
使用多线程访问一个资源,这个资源始终都能表现出正确的行为。
不被运行的环境影响、多线程可以交替访问、不需要任何额外的同步和协同。
2、Java实现多线程安全生产者
这里只是模拟多线程环境下使用生产者发送消息,其实没有做额外的线程安全操作,就是把生产者当成了一个公共资源,所有线程都可以访问这个生产者。
kafka默认客户端提供的生产者本身就是线程安全的,因为生产者发送消息只有一步操作,就是发送消息。只要消息进入消息缓冲区就可以发送给broker,不会出现消息重复发送。
package com.allwe.client.concurrent;
import com.allwe.client.partitioner.MyPartitioner;
import lombok.Data;
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.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 生产者多线程安全 - 测试demo
*
* @Author: AllWe
* @Date: 2024/09/27/9:30
*/
@Data
@Slf4j
public class ConcurrentProducerWorker {
/**
* 消息数量
*/
private static final int RECORD_COUNT = 1000;
/**
* 固定线程池 - 线程数等于CPU核数
*/
private static final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
/**
* 发令枪
*/
private static final CountDownLatch countDownLatch = new CountDownLatch(RECORD_COUNT);
/**
* 生产者 - 这里让所有的线程都共享同一个生产者
*/
private static KafkaProducer<String, String> kafkaProducer;
/**
* 类初始化的时候 - 创建生产者实例
*/
static {
// 设置属性
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);
kafkaProducer = new KafkaProducer<>(properties);
}
/**
* 启动器
*/
public static void main(String[] args) {
try {
// 循环创建消息
for (int count = 0; count < RECORD_COUNT; count++) {
ProducerRecord<String, String> record = new ProducerRecord<>("topic_6", "allwe", "allwe_" + count);
executorService.submit(new ConcurrentProducer(record, kafkaProducer, countDownLatch));
}
countDownLatch.await();
} catch (Exception e) {
e.printStackTrace();
} finally {
// 关闭生产者连接
kafkaProducer.close();
// 释放线程池资源
executorService.shutdown();
}
}
}
package com.allwe.client.concurrent;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.concurrent.CountDownLatch;
/**
* 生产者多线程安全 - 测试demo
*
* @Author: AllWe
* @Date: 2024/09/27/9:30
*/
@Data
@Slf4j
public class ConcurrentProducer implements Runnable {
/**
* 消息体
*/
private ProducerRecord<String, String> record;
/**
* 生产者
*/
private KafkaProducer<String, String> producer;
/**
* 发令枪
*/
private CountDownLatch countDownLatch;
public ConcurrentProducer(ProducerRecord<String, String> record, KafkaProducer<String, String> producer, CountDownLatch countDownLatch) {
this.record = record;
this.producer = producer;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
try {
String name = Thread.currentThread().getName();
producer.send(record, new ConcurrentCallBackImpl(name));
countDownLatch.countDown();
} catch (Exception e) {
e.printStackTrace();
}
}
}
package com.allwe.client.concurrent;
import cn.hutool.core.util.ObjectUtil;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
* 异步发送消息回调解析器
*
* @Author: AllWe
* @Date: 2024/09/27/9:30
*/
public class ConcurrentCallBackImpl implements Callback {
private String threadName;
public ConcurrentCallBackImpl(String threadName) {
this.threadName = threadName;
}
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (ObjectUtil.isNull(e)) {
// 解析回调元数据
System.out.println(threadName + "|-offset:" + recordMetadata.offset() + ",partition:" + recordMetadata.partition());
} else {
e.printStackTrace();
}
}
}
3、Java实现多线程安全消费者
kafka客户端提供的消费者不是多线程安全的,是因为消费者在消费消息的时候,需要有2步操作:取消息和ACK确认,在多线程场景下可能会出现:
① 线程1取到了消息,但是没来得及进行ACK确认。
② 线程2进来了,又消费了一次相同的消息。
③ 线程2提交ACK确认。
④ 线程1提交ACK确认。
这样就会产生重复消费,这个时候就需要对消费者进行额外处理。
有两个处理方案:
① 给消费过程加锁,但是会降低程序执行效率。
② 每一个线程都创建自己的消费者,只消费自己分区内的数据。
我写的demo是使用第二种办法。
package com.allwe.client.concurrent;
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 java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
* 线程安全消费者 - 测试demo
*
* @Author: AllWe
* @Date: 2024/09/27/12:19
*/
@Data
@Slf4j
public class ConcurrentConsumer implements Runnable {
/**
* 消费者配置参数
*/
private Properties properties;
/**
* 群组id
*/
private String groupId;
/**
* 消费主题
*/
private String topicName;
/**
* 消费者实例
*/
private KafkaConsumer<String, String> consumer;
public ConcurrentConsumer(Properties properties, String groupId, String topicName) {
this.properties = properties;
this.groupId = groupId;
this.topicName = topicName;
// 补充配置参数
properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
// 创建消费者实例 - 每一个线程都创建自己的消费者,避免共享相同的消费者实例
consumer = new KafkaConsumer<>(properties);
// 配置消费主题
consumer.subscribe(Collections.singleton(topicName));
}
@Override
public void run() {
try {
String threadName = Thread.currentThread().getName();
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> record : records) {
StringBuilder stringBuilder = new StringBuilder(threadName).append("|-");
stringBuilder.append("partition:").append(record.partition());
stringBuilder.append("offset:").append(record.offset());
stringBuilder.append("key:").append(record.key());
stringBuilder.append("value:").append(record.value());
System.out.println(stringBuilder);
}
}
} finally {
consumer.close();
}
}
}
package com.allwe.client.concurrent;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 多线程安全消费者 - 测试demo
*
* @Author: AllWe
* @Date: 2024/09/27/12:34
*/
@Data
@Slf4j
public class ConcurrentConsumerWorker {
/**
* 消费线程数
*/
private static final Integer THREAD_COUNT = 2;
/**
* 线程池 - 2个线程,别超过目标主题的分区数
*/
private static ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
public static void main(String[] args) {
// 消费者配置
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
properties.put("key.deserializer", StringDeserializer.class);
properties.put("value.deserializer", StringDeserializer.class);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从头开始消费
for (Integer i = 0; i < THREAD_COUNT; i++) {
executorService.submit(new ConcurrentConsumer(properties, "allwe01", "topic_6"));
}
}
}
六、群组协调
1、群主
在每一个群组内部,都有一个【群主】。往往是第一个注册进入群组的消费者承担,它的职责是读取当前群组消费的主题,以及目标主题的分区信息。
群主节点的数据权限高于普通消费者,它可以获取全部消费者节点对应的分区信息。但是普通消费者节点只能看见本节点的分区信息。
2、消费者协调器
属于客户端,每个消费者群组内部都有一个消费者协调器,用于获取群主节点保存的分区信息,再协调群组内的其他消费者处理哪些主题和分区。
分配好任务后将配置信息推送给【组协调器】,组协调器再将消息发送给不同的消费者。
当群组内出现某个节点掉线、上线时,消费者协调器也会参与协调。
1、向【组协调器】发送入组请求。
2、发起同步组的请求 -- 由群组计算分配策略,确定消费者的分区划分,发送给组协调器。
3、心跳机制(与组协调器维持)。
4、提交ACK确认(发起已经提交的消费偏移量的请求)。
5、主动发起离组请求。
3、组协调器
属于kafka broker,主要负责以下功能:
1、处理申请加入群组的消费者,并且选举群主。
2、收到同步组的请求后,触发分区再均衡,同步新的分配方案。
3、心跳机制(与客户端维持),如果得知哪些客户端掉线了,触发分区再均衡机制。
4、管理消费者已经消费的偏移量,保存在主题【__consumer_offsets】,默认有50个分区。
4、新的消费者加入群组的处理流程
1、消费者客户端启动、重连,都会给组协调器发送一个入组请求(joinGroup请求)。
2、消费者客户端完成joinGroup后,消费者协调器向组协调器发起同步组请求(SyncGroup请求),获取新的分配方案。
3、入组后保持心跳(客户端控制参数:max.poll.interval.ms)。
4、消费者客户端掉线,触发离组处理。
5、消费者群组的信息存储在哪里
存储在__consumer_offsets文件中,groupName.hashCode() % 50,获取配置文件的编号。
七、分区再均衡
1、功能
针对单个消费者群组,对群组内的消费者负责的分区进行重新分配。
1、假设【主题α】有三个分区,分别是①、②、③。
2、进来两个消费者A、B。A负责分区①,B负责分区②③。
3、又进来一个消费者C,再均衡监听器就把分区③分配给C。
4、消费者C掉线,再均衡监听器把分区③分配给A或者B。
2、Java代码验证分区再均衡
package com.allwe.client.reBalance;
import lombok.Data;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 分区再均衡处理器
*
* @Author: AllWe
* @Date: 2024/10/17/8:05
*/
@Data
public class ReBalanceHandler implements ConsumerRebalanceListener {
// 记录每个分区的消费偏移量
public final static ConcurrentHashMap<TopicPartition, Long> partitionOffsetMap = new ConcurrentHashMap<TopicPartition, Long>();
private final Map<TopicPartition, OffsetAndMetadata> currOffsets;
private final KafkaConsumer<String, String> consumer;
public ReBalanceHandler(Map<TopicPartition, OffsetAndMetadata> currOffsets, KafkaConsumer<String, String> consumer) {
this.currOffsets = currOffsets;
this.consumer = consumer;
}
// 分区再均衡之前
// 某一个消费者在让出分区之前,需要先将已消费的偏移量提交
@Override
public void onPartitionsRevoked(Collection<TopicPartition> collection) {
// 线程id
final String id = Thread.currentThread().getId() + "";
System.out.println(id + "-onPartitionsRevoked参数值为:" + collection);
System.out.println(id + "-服务器准备分区再均衡,提交偏移量。当前偏移量为:" + currOffsets);
//我们可以不使用consumer.commitSync(currOffsets);
//提交偏移量到kafka,由我们自己维护*/
//开始事务
//偏移量写入数据库
System.out.println("分区偏移量表中:" + partitionOffsetMap);
for (TopicPartition topicPartition : collection) {
partitionOffsetMap.put(topicPartition, currOffsets.get(topicPartition).offset());
}
// 同步提交偏移量,等到成功后再往后执行
consumer.commitSync(currOffsets);
}
// 分区再均衡之后
// 新的消费者接管分区后,从上一次的偏移量开始消费
@Override
public void onPartitionsAssigned(Collection<TopicPartition> collection) {
// 线程id
final String threadId = Thread.currentThread().getId() + "";
System.out.println(threadId + "|-再均衡完成,onPartitionsAssigned参数值为:" + collection);
System.out.println("分区偏移量表中:" + partitionOffsetMap);
for (TopicPartition topicPartition : collection) {
System.out.println(threadId + "-topicPartition" + topicPartition);
// 取得接管分区之前的偏移量
Long offset = partitionOffsetMap.get(topicPartition);
if (offset == null) continue;
consumer.seek(topicPartition, partitionOffsetMap.get(topicPartition));
}
}
@Override
public void onPartitionsLost(Collection<TopicPartition> partitions) {
ConsumerRebalanceListener.super.onPartitionsLost(partitions);
}
}
package com.allwe.client.reBalance;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
/**
* 线程安全消费者 - 测试demo
*
* @Author: AllWe
* @Date: 2024/09/27/12:19
*/
@Data
@Slf4j
public class ConcurrentConsumer implements Runnable {
/**
* 消费者配置参数
*/
private Properties properties;
/**
* 群组id
*/
private String groupId;
/**
* 消费主题
*/
private String topicName;
/**
* 消费者实例
*/
private KafkaConsumer<String, String> consumer;
/**
* 记录分区消费者偏移量
*/
private final Map<TopicPartition, OffsetAndMetadata> currOffsets = new HashMap<>();
public ConcurrentConsumer(Properties properties, String groupId, String topicName) {
this.properties = properties;
this.groupId = groupId;
this.topicName = topicName;
// 补充配置参数
properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
// 创建消费者实例 - 每一个线程都创建自己的消费者,避免共享相同的消费者实例
consumer = new KafkaConsumer<>(properties);
// 配置消费主题 - 配置再均衡监听器
consumer.subscribe(Collections.singleton(topicName), new ReBalanceHandler(currOffsets,consumer));
}
@Override
public void run() {
try {
String threadName = Thread.currentThread().getName();
Integer offset = 0;
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> record : records) {
StringBuilder stringBuilder = new StringBuilder(threadName).append("|-");
stringBuilder.append("partition:").append(record.partition());
stringBuilder.append(",offset:").append(record.offset());
stringBuilder.append(",key:").append(record.key());
stringBuilder.append(",value:").append(record.value());
System.out.println(stringBuilder);
offset++;
currOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(offset, "no"));
}
}
} finally {
consumer.close();
}
}
}
自定义一个再均衡监听器,消费者在订阅接口中指定这个监听器,即可自动执行监听器的任务。
// 配置消费主题 - 配置再均衡监听器
consumer.subscribe(Collections.singleton(topicName), new ReBalanceHandler(currOffsets,consumer));
版权归原作者 AllWe$ 所有, 如有侵权,请联系我们删除。