SpringBoot整合Kafka工具类的详细代码
kafka是什么?
Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者在网站中的所有动作流数据。 这种动作(网页浏览,搜索和其他用户的行动)是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统,但又要求实时处理的限制,这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理,也是为了通过集群来提供实时的消息。
应用场景
- 消息系统: Kafka 和传统的消息系统(也称作消息中间件)都具备系统解耦、冗余存储、流量削峰、缓冲、异步通信、扩展性、可恢复性等功能。与此同时,Kafka 还提供了大多数消息系统难以实现的消息顺序性保障及回溯消费的功能。
- 存储系统: Kafka 把消息持久化到磁盘,相比于其他基于内存存储的系统而言,有效地降低了数据丢失的风险。也正是得益于 Kafka 的消息持久化功能和多副本机制,我们可以把 Kafka 作为长期的数据存储系统来使用,只需要把对应的数据保留策略设置为“永久”或启用主题的日志压缩功能即可。
- 流式处理平台: Kafka 不仅为每个流行的流式处理框架提供了可靠的数据来源,还提供了一个完整的流式处理类库,比如窗口、连接、变换和聚合等各类操作。
下面看下SpringBoot整合Kafka工具类的详细代码。
pom.xml
<dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.12.0</version> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.6.3</version> </dependency> <dependency> <groupId>fastjson</groupId> <artifactId>fastjson</artifactId> <version>1.2.83</version> </dependency>
工具类
package com.bbl.demo.utils; import org.apache.commons.lang3.exception.ExceptionUtils; import org.apache.kafka.clients.admin.*; 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.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.errors.TopicExistsException; import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; import com.alibaba.fastjson.JSONObject; import java.time.Duration; import java.util.*; import java.util.concurrent.ExecutionException; public class KafkaUtils { private static AdminClient admin; /** * 私有静态方法,创建Kafka生产者 * @author o * @return KafkaProducer */ private static KafkaProducer<String, String> createProducer() { Properties props = new Properties(); //声明kafka的地址 props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node01:9092,node02:9092,node03:9092"); //0、1 和 all:0表示只要把消息发送出去就返回成功;1表示只要Leader收到消息就返回成功;all表示所有副本都写入数据成功才算成功 props.put("acks", "all"); //重试次数 props.put("retries", Integer.MAX_VALUE); //批处理的字节数 props.put("batch.size", 16384); //批处理的延迟时间,当批次数据未满之时等待的时间 props.put("linger.ms", 1); //用来约束KafkaProducer能够使用的内存缓冲的大小的,默认值32MB props.put("buffer.memory", 33554432); // properties.put("value.serializer", // "org.apache.kafka.common.serialization.ByteArraySerializer"); // properties.put("key.serializer", // "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); return new KafkaProducer<String, String>(props); } /** * 私有静态方法,创建Kafka消费者 * @author o * @return KafkaConsumer */ private static KafkaConsumer<String, String> createConsumer() { Properties props = new Properties(); //声明kafka的地址 props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node01:9092,node02:9092,node03:9092"); //每个消费者分配独立的消费者组编号 props.put("group.id", "111"); //如果value合法,则自动提交偏移量 props.put("enable.auto.commit", "true"); //设置多久一次更新被消费消息的偏移量 props.put("auto.commit.interval.ms", "1000"); //设置会话响应的时间,超过这个时间kafka可以选择放弃消费或者消费下一条消息 props.put("session.timeout.ms", "30000"); //自动重置offset props.put("auto.offset.reset","earliest"); // properties.put("value.serializer", // "org.apache.kafka.common.serialization.ByteArraySerializer"); // properties.put("key.serializer", // "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); return new KafkaConsumer<String, String>(props); } /** * 私有静态方法,创建Kafka集群管理员对象 * @author o */ public static void createAdmin(String servers){ Properties props = new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,servers); admin = AdminClient.create(props); } /** * 私有静态方法,创建Kafka集群管理员对象 * @author o * @return AdminClient */ private static void createAdmin(){ createAdmin("node01:9092,node02:9092,node03:9092"); } /** * 传入kafka约定的topic,json格式字符串,发送给kafka集群 * @author o * @param topic * @param jsonMessage */ public static void sendMessage(String topic, String jsonMessage) { KafkaProducer<String, String> producer = createProducer(); producer.send(new ProducerRecord<String, String>(topic, jsonMessage)); producer.close(); } /** * 传入kafka约定的topic消费数据,用于测试,数据最终会输出到控制台上 * @author o * @param topic */ public static void consume(String topic) { KafkaConsumer<String, String> consumer = createConsumer(); consumer.subscribe(Arrays.asList(topic)); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100)); for (ConsumerRecord<String, String> record : records){ System.out.printf("offset = %d, key = %s, value = %s",record.offset(), record.key(), record.value()); System.out.println(); } } } /** * 传入kafka约定的topic数组,消费数据 * @author o * @param topics */ public static void consume(String ... topics) { KafkaConsumer<String, String> consumer = createConsumer(); consumer.subscribe(Arrays.asList(topics)); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100)); for (ConsumerRecord<String, String> record : records){ System.out.printf("offset = %d, key = %s, value = %s",record.offset(), record.key(), record.value()); System.out.println(); } } } /** * 传入kafka约定的topic,json格式字符串数组,发送给kafka集群 * 用于批量发送消息,性能较高。 * @author o * @param topic * @param jsonMessages * @throws InterruptedException */ public static void sendMessage(String topic, String... jsonMessages) throws InterruptedException { KafkaProducer<String, String> producer = createProducer(); for (String jsonMessage : jsonMessages) { producer.send(new ProducerRecord<String, String>(topic, jsonMessage)); } producer.close(); } /** * 传入kafka约定的topic,Map集合,内部转为json发送给kafka集群 <br> * 用于批量发送消息,性能较高。 * @author o * @param topic * @param mapMessageToJSONForArray */ public static void sendMessage(String topic, List<Map<Object, Object>> mapMessageToJSONForArray) { KafkaProducer<String, String> producer = createProducer(); for (Map<Object, Object> mapMessageToJSON : mapMessageToJSONForArray) { String array = JSONObject.toJSON(mapMessageToJSON).toString(); producer.send(new ProducerRecord<String, String>(topic, array)); } producer.close(); } /** * 传入kafka约定的topic,Map,内部转为json发送给kafka集群 * @author o * @param topic * @param mapMessageToJSON */ public static void sendMessage(String topic, Map<Object, Object> mapMessageToJSON) { KafkaProducer<String, String> producer = createProducer(); String array = JSONObject.toJSON(mapMessageToJSON).toString(); producer.send(new ProducerRecord<String, String>(topic, array)); producer.close(); } /** * 创建主题 * @author o * @param name 主题的名称 * @param numPartitions 主题的分区数 * @param replicationFactor 主题的每个分区的副本因子 */ public static void createTopic(String name,int numPartitions,int replicationFactor){ if(admin == null) { createAdmin(); } Map<String, String> configs = new HashMap<>(); CreateTopicsResult result = admin.createTopics(Arrays.asList(new NewTopic(name, numPartitions, (short) replicationFactor).configs(configs))); //以下内容用于判断创建主题的结果 for (Map.Entry<String, KafkaFuture<Void>> entry : result.values().entrySet()) { try { entry.getValue().get(); System.out.println("topic "+entry.getKey()+" created"); } catch (InterruptedException | ExecutionException e) { if (ExceptionUtils.getRootCause(e) instanceof TopicExistsException) { System.out.println("topic "+entry.getKey()+" existed"); } } } } /** * 删除主题 * @author o * @param names 主题的名称 */ public static void deleteTopic(String name,String ... names){ if(admin == null) { createAdmin(); } Map<String, String> configs = new HashMap<>(); Collection<String> topics = Arrays.asList(names); topics.add(name); DeleteTopicsResult result = admin.deleteTopics(topics); //以下内容用于判断删除主题的结果 for (Map.Entry<String, KafkaFuture<Void>> entry : result.values().entrySet()) { try { entry.getValue().get(); System.out.println("topic "+entry.getKey()+" deleted"); } catch (InterruptedException | ExecutionException e) { if (ExceptionUtils.getRootCause(e) instanceof UnknownTopicOrPartitionException) { System.out.println("topic "+entry.getKey()+" not exist"); } } } } /** * 查看主题详情 * @author o * @param names 主题的名称 */ public static void describeTopic(String name,String ... names){ if(admin == null) { createAdmin(); } Map<String, String> configs = new HashMap<>(); Collection<String> topics = Arrays.asList(names); topics.add(name); DescribeTopicsResult result = admin.describeTopics(topics); //以下内容用于显示主题详情的结果 for (Map.Entry<String, KafkaFuture<TopicDescription>> entry : result.values().entrySet()) { try { entry.getValue().get(); System.out.println("topic "+entry.getKey()+" describe"); System.out.println("\t name: "+entry.getValue().get().name()); System.out.println("\t partitions: "); entry.getValue().get().partitions().stream().forEach(p-> { System.out.println("\t\t index: "+p.partition()); System.out.println("\t\t\t leader: "+p.leader()); System.out.println("\t\t\t replicas: "+p.replicas()); System.out.println("\t\t\t isr: "+p.isr()); }); System.out.println("\t internal: "+entry.getValue().get().isInternal()); } catch (InterruptedException | ExecutionException e) { if (ExceptionUtils.getRootCause(e) instanceof UnknownTopicOrPartitionException) { System.out.println("topic "+entry.getKey()+" not exist"); } } } } /** * 查看主题列表 * @author o * @return Set<String> TopicList */ public static Set<String> listTopic(){ if(admin == null) { createAdmin(); } ListTopicsResult result = admin.listTopics(); try { result.names().get().stream().map(x->x+"\t").forEach(System.out::print); return result.names().get(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); return null; } } public static void main(String[] args) { System.out.println(listTopic()); } }
到此这篇关于SpringBoot整合Kafka工具类的文章就介绍到这了,更多相关SpringBoot整合Kafka工具类内容请搜索猪先飞以前的文章或继续浏览下面的相关文章希望大家以后多多支持猪先飞!
原文出处:https://www.cnblogs.com/big-keyboard/p/16731108.html
相关文章
解决springboot使用logback日志出现LOG_PATH_IS_UNDEFINED文件夹的问题
这篇文章主要介绍了解决springboot使用logback日志出现LOG_PATH_IS_UNDEFINED文件夹的问题,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-04-28- 这篇文章主要介绍了.NET Core下使用Kafka的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-09-22
- 这篇文章主要为大家详细介绍了SpringBoot实现excel文件生成和下载,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2021-02-09
- 这篇文章主要介绍了详解springBoot启动时找不到或无法加载主类解决办法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-09-16
- 这篇文章主要介绍了SpringBoot集成Redis实现消息队列的方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-02-10
- 这篇文章主要介绍了解决Springboot get请求是参数过长的情况,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-09-17
Spring Boot项目@RestController使用重定向redirect方式
这篇文章主要介绍了Spring Boot项目@RestController使用重定向redirect方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-02- 这篇文章主要介绍了Springboot+TCP监听服务器搭建过程,本文通过图文并茂的形式给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2020-10-28
- 这篇文章主要介绍了springBoot 项目排除数据库启动方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-10
详解SpringBoot之访问静态资源(webapp...)
这篇文章主要介绍了详解SpringBoot之访问静态资源(webapp...),文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-09-14- 这篇文章主要介绍了SpringBoot接口接收json参数解析,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-10-19
springboot中使用@Transactional注解事物不生效的坑
这篇文章主要介绍了springboot中使用@Transactional注解事物不生效的原因,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-01-26- 这篇文章主要介绍了springboot多模块包扫描问题的解决方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-09-16
Springboot+MDC+traceId日志中打印唯一traceId
本文主要介绍了Springboot+MDC+traceId日志中打印唯一traceId,文中通过示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2021-10-17Springboot mybatis plus druid多数据源解决方案 dynamic-datasource的使用详解
这篇文章主要介绍了Springboot mybatis plus druid多数据源解决方案 dynamic-datasource的使用,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2020-11-18- 这篇文章主要介绍了Springboot实现多线程注入bean的工具类操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-08-27
SpringBoot部署到Linux读取resources下的文件及遇到的坑
本文主要给大家介绍SpringBoot部署到Linux读取resources下的文件,在平时业务开发过程中,很多朋友在获取到文件内容乱码或者文件读取不到的问题,今天给大家分享小编遇到的坑及处理方案,感兴趣的朋友跟随小编一起看看吧...2021-06-21- 这篇文章主要介绍了springboot中nacos动态路由的配置方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-11
SpringBoot高版本修改为低版本时测试类报错的解决方案
这篇文章主要介绍了SpringBoot高版本修改为低版本时测试类报错的解决方案,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-18解决Springboot整合shiro时静态资源被拦截的问题
这篇文章主要介绍了解决Springboot整合shiro时静态资源被拦截的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-01-26