SpringBoot整合Kafka工具类的详细代码
作者:键盘命 发布时间:2022-07-03 14:12:13
标签: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());
}
}
来源:https://www.cnblogs.com/big-keyboard/p/16731108.html


猜你喜欢
- 前言众所周知,RxJava2 中当链式调用中抛出异常时,如果没有对应的 Consumer 去处理异常,则这个异常会被抛出到虚拟机中去,And
- 上下文:程序运行需要的环境(外部变量)上下文切换:将之前的程序需要的外部变量复制保存,然后切换到新的程序运行环境系统调用:(用户态陷入操作系
- 本文实例为大家分享了C#异步调用的具体代码,供大家参考,具体内容如下using System;using System.Collection
- 一、介绍这个类是继承自ImageView的,所以对于这个控件我们可以使用ImageView的所有属性android.support.desi
- 1. 基础知识集合Java.util包下的常用子类,集合无非就是各种数据结构的应用。集合存在的目的就是为了将数据高效的进行读写,无论哪种具体
- C#编程语言非常优美,我个人还是非常赞同的。特别是在学习一段时间C#后发现确实在它的语法和美观度来说确实要比其它编程语言强一些(也可能是由于
- 微服务feign调用添加token1.一般情况是这么配置的具体的怎么调用就不说了 如下配置,就可以在请求头中添加需要的请求头信息。packa
- package com.action.android_test;import android.location.Location;impor
- Java常用类String类概述String类:代表字符串。Java程序中的所有字符串字面值(如:”abc“)都作为子类的实例实现Strin
- 平时项目中只要涉及表,那么一定能接触到众多各式各样的ID编号,博主整理一些常用的ID格式,整合一个ID生成工具类,供大家参考,如果有什么不足
- 本文实例分析了C#实现的24点游戏。分享给大家供大家参考。具体如下:1. 24点游戏规则及算法规则:给出4个自然数,找出能够求出24的四则运
- 一、前言通过前面我们也知道,通过getMapper方式来进行查询,最后会通过mapperMehod类,对接口中传来的参数也会在这个类里面进行
- 一、介绍SharpZipLib是一个完全由C#编写的ZIP,GZIP,Tar和BZIP2 Library,可以方便的支持这几种格式的压缩和解
- 1. 实验目的: 使用线程池的时候,有时候需要考虑服务器的最大线程数目和程序最快
- 目录基于Java的guava开源库工具类1、guava的maven配置引入 2、LoadingCache3、Multimap 和
- 本文是利用C# 实现FTP客户端的小例子,主要实现上传,下载,删除等功能,以供学习分享使用。思路:通过读取FTP站点的目录信息,列出对应的文
- 抽象类往往用来表征对问题领域进行分析、设计中得出的抽象概念,是对一系列看上去不同,但是本质上相同的具体概念的抽象。下面我们以水果为例,首先定
- 什么是事务处理事务是计算机应用中不可或缺的组件模型,它保证了用户操作的原子性 ( Atomicity )、一致性 ( Consistency
- 使用Task类解决线程的等待问题在任何的编程语言中,面对耗时任务时,我们都会有这样的需求:让任务执行一定时间,主任务进行等待,如果到时仍然完
- 一款书籍阅读器,需要以下功能才能说的上比较完整:文字页面展示,即书页;页面之间的跳转动画,即翻页动作;能够在每一页上记录阅读进度,即书签;能