kafka第二次课！！！

package com.aaa.kafka.util;
import org.apache.kafka.clients.admin.;
import org.apache.kafka.common.KafkaFuture;
import java.util.;
import java.util.concurrent.ExecutionException;
/**
• @ fileName:KafkaJavaUtil
• @ description:kafka通用工具类
• @ author:zhz
• @ createTime:2022/1/11 10:24
• @ version:1.0.0/
public class KafkaJavaUtil {
//多个方法公共，写为类属性
private AdminClient adminClient;
/*
• 使用空构造，初始化AdminClient
/
public KafkaJavaUtil() {
//加载配置
Properties properties = LoadConfigurationUtil.loadProperty(“kafka.properties”);
//使用配置文件初始化对象
adminClient = KafkaAdminClient.create(properties);
}
/*
• 获取topic列表
• @return
/
public List getTopicList(){
try {
//列表选项
ListTopicsOptions listTopicsOptions =new ListTopicsOptions();
//设置是否显示kafka内部使用的topic //listInternal(true) __consumer_offsets 会显示 否则不会
listTopicsOptions.listInternal(false);
//返回topic列表
ListTopicsResult listTopicsResult = adminClient.listTopics(listTopicsOptions);
//名称集合对象
KafkaFuture nameList = listTopicsResult.names();
//获取到set集合
Set setString = nameList.get();
//定义List
List topicNameList = new ArrayList<>(setString);
return topicNameList;
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}finally {
if(adminClient!=null){
adminClient.close();
}
}
return null;
}
/*
• 创建topic
• @param topicName 名称
• @param partitions 分区数量
• @param replicationFactor 副本因子
• @return
/
public boolean createTopic(String topicName,int partitions,short replicationFactor){
try {
NewTopic topic =new NewTopic(topicName,partitions,replicationFactor);
adminClient.createTopics(Arrays.asList(topic));
//无异常创建成功 返回true
return true;
} catch (Exception e) {
e.printStackTrace();
}finally {
if(adminClient!=null){
adminClient.close();
}
}
return false;
}
/*
• 删除topic
• @param topicName
• @return
*/
public boolean deleteTopic(String topicName){
try {
adminClient.deleteTopics(Arrays.asList(topicName));
return true;
} catch (Exception e) {
e.printStackTrace();
}finally {
if(adminClient!=null){
adminClient.close();
}
}
return false;
}
}
KafkaJavaUtilTest：
package com.aaa.kafka.test;
import com.aaa.kafka.util.KafkaJavaUtil;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.util.List;
import static org.junit.Assert.*;
public class KafkaJavaUtilTest {
private KafkaJavaUtil kafkaJavaUtil;
@Before
public void setUp() throws Exception {
kafkaJavaUtil = new KafkaJavaUtil();
}

package com.aaa.kafka.demo;
import com.aaa.kafka.util.LoadConfigurationUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
• @ fileName:ProducerKafka
• @ description:
• @ author:zhz
• @ createTime:2022/1/12 10:10
• @ version:1.0.0
*/
public class ProducerKafka {
public static void main(String[] args) {
KafkaProducer kafkaProducer = null;
try {
//加载配置文件并生成Properties
Properties properties = LoadConfigurationUtil.loadProperty(“producer.properties”);
//实例化KafkaProducer
kafkaProducer = new KafkaProducer(properties);
for (int i = 0; i < 100; i++) {
System.out.println(“发送了信息hello”+i);
ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),“hello”+i);
//发送消息
kafkaProducer.send(producerRecord);
//每个200毫秒发送一次
Thread.sleep(200);
}
System.out.println(“消息发送完毕！！！”);
} catch (Exception e) {
e.printStackTrace();
} finally {
if(kafkaProducer!=null){
kafkaProducer.close();
}
}
}
}
带回调的生产者代码类
package com.aaa.kafka.demo;
import com.aaa.kafka.util.LoadConfigurationUtil;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Properties;
/**
• @ fileName:ProducerKafka
• @ description:
• @ author:zhz
• @ createTime:2022/1/12 10:10
• @ version:1.0.0
*/
public class ProducerKafkaWithCallback {
public static void main(String[] args) {
KafkaProducer kafkaProducer = null;
try {
//加载配置文件并生成Properties
Properties properties = LoadConfigurationUtil.loadProperty(“producer.properties”);
//实例化KafkaProducer
kafkaProducer = new KafkaProducer(properties);
for (int i = 0; i < 100; i++) {
System.out.println(“发送了信息hi,AAA”+i+“发送后的信息为：”);
//3，不指定分区，也没有key
//ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),“hello,qy141”+i);
//2,不指定分区，但是有key
ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),“AAA”+i,“hi,AAA”+i);
//1,指定分区,跟key和value无关
//ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),2,“key”+i,“hi”+i);
//发送消息,并带回调函数（根据回调函数可以看到该消息被存放分区和偏移量等信息）
kafkaProducer.send(producerRecord, new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
System.out.println(“-------该消息的分区为：”+recordMetadata.partition()+“,偏移量为：”+recordMetadata.offset());
}
});
//每个200毫秒发送一次
Thread.sleep(200);
}
System.out.println(“消息发送完毕！！！”);
} catch (Exception e) {
e.printStackTrace();
} finally {
if(kafkaProducer!=null){
kafkaProducer.close();
}
}
}
}

带回调的生产者代码类
package com.aaa.kafka.demo;
import com.aaa.kafka.util.LoadConfigurationUtil;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Properties;
/**
• @ fileName:ProducerKafka
• @ description:
• @ author:zhz
• @ createTime:2022/1/12 10:10
• @ version:1.0.0
*/
public class ProducerKafkaWithCallback {
public static void main(String[] args) {
KafkaProducer kafkaProducer = null;
try {
//加载配置文件并生成Properties
Properties properties = LoadConfigurationUtil.loadProperty(“producer.properties”);
//实例化KafkaProducer
kafkaProducer = new KafkaProducer(properties);
for (int i = 0; i < 100; i++) {
System.out.println(“发送了信息hi,AAA”+i+“发送后的信息为：”);
//3，不指定分区，也没有key
//ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),“hello,qy141”+i);
//2,不指定分区，但是有key
ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),“AAA”+i,“hi,AAA”+i);
//1,指定分区,跟key和value无关
//ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),2,“key”+i,“hi”+i);
//发送消息,并带回调函数（根据回调函数可以看到该消息被存放分区和偏移量等信息）
kafkaProducer.send(producerRecord, new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
System.out.println(“-------该消息的分区为：”+recordMetadata.partition()+“,偏移量为：”+recordMetadata.offset());
}
});
//每个200毫秒发送一次
Thread.sleep(200);
}
System.out.println(“消息发送完毕！！！”);
} catch (Exception e) {
e.printStackTrace();
} finally {
if(kafkaProducer!=null){
kafkaProducer.close();
}
}
}
}

#anything else: throw exception to the consumer.直接给消费者抛出异常
auto.offset.reset=earliest
#自定义指定消费topic名称
topicName=tp88
消费者类：
package com.aaa.kafka.demo;
import com.aaa.kafka.util.LoadConfigurationUtil;
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.time.temporal.TemporalUnit;
import java.util.Arrays;
import java.util.Properties;
/**
• @ fileName:ConsumerKafka
• @ description:消息消费者
• @ author:zhz
• @ createTime:2022/1/12 9:30
• @ version:1.0.0
*/
public class ConsumerKafka {
public static void main(String[] args) {
KafkaConsumer kafkaConsumer = null;
try {
//加载配置文件并生成Properties
Properties properties = LoadConfigurationUtil.loadProperty(“consumer.properties”);
//properties.put(“bootstrap.servers”,“cluster1:9092,cluster2:9092,cluster3:9092”);
//实例化kafkaConsumer
kafkaConsumer = new KafkaConsumer(properties);
//指定当前消费者订阅的topic列表 可以是多个
kafkaConsumer.subscribe(Arrays.asList(properties.getProperty(“topicName”)));
//添加提示
System.out.println(“消费者已经启动，等待生产者生产消息。。。。。。”);
//编写循环接受消息
while (true){
//如果说topic中暂时没有生成者生产消息，就会隔间500毫秒拉取一次
ConsumerRecords consumerRecords = kafkaConsumer.poll(500);
//循环拉取集合结果
for (ConsumerRecord consumerRecord : consumerRecords) {
System.out.println(“消息的key：”+consumerRecord.key()+“，value:”+consumerRecord.value()
+“，offset偏移量：”+consumerRecord.offset()+“,partition消息所在分区：”+consumerRecord.partition());
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
//关闭资源
if(kafkaConsumer!=null){
kafkaConsumer.close();
}
}
}
}