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
0
投稿猜你喜欢
1. 背景在业务处理完之后，需要调用其他系统的接口，将相应的处理结果通知给对方，若是同步请求，假如调用的系统出现异常或是宕机等事件，会导致自- 1 简介Solace是一个强大的实时性的事件驱动消息队列。本文将介绍如何在Spring中使用，虽然代码使用的是Spring Boot，但并没
- GlobalLock的作用对于某条数据进行更新操作，如果全局事务正在进行，当某个本地事务需要更新该数据时，需要使用@GlobalLock确保
- 本文实例讲述了Java解析Excel内容的方法。分享给大家供大家参考。具体实现方法如下：import java.io.File;
- MyBatis核心配置文件<?xml version="1.0" encoding="UTF-8&quo
- SpringBoot的持久化层可以是Spring内置的轻量级JdbcTemplate、也可以是Hibernate或Mybatis等等，只需要
- ArrayList和LinkedList都实现了List接口，有以下的不同点：1、ArrayList是基于索引的数据接口，它的底层是数组。它
- java 获取字节码文件的几种方法总结在本文中，以Person类为例，将分别演示获取该类字节码文件的三种方式，其具体思想及代码如下所示：pu
- 本文主要介绍在SpringMVC框架整合Junit框架进行单元测试。闲话少述，让我们直入主题。系统环境软件版本spring-webmvc4.
- 本文实例讲述了C#中HttpWebRequest的用法。分享给大家供大家参考。具体如下：HttpWebRequest类主要利用HTTP 协议
- 一个非侵入的api编译、收集、Rest文档生成工具。工具通过分析代码和注释，获取文档信息，生成RestDoc文档前言程序员一直以来都有一个烦
- 1、数组理论基础数组是存放在连续内存空间上的相同类型数据的集合，可以通过下标索引的方式获取到下标下对应的数据。举个栗子（字符数组）~可以看到
- java8的stream流能完美解对象集合去重问题. List<UserCar> list1 = new ArrayList()
- 持久层的那些事什么是 JDBCJDBC（JavaDataBase Connectivity）就是 Java 数据库连接, 说的直白点就是 使
- 写在前面之前想尝试把JWT和Shiro结合到一起，但是在网上查了些博客，也没太有看懂，所以就自己重新研究了一下Shiro的工作机制，然后自己
- 本文实例讲述了Java循环队列原理与用法。分享给大家供大家参考，具体如下：在正式进行循环队列学习之前，我们先来看看在顺序队列中删除队首元素出
- 会话技术会话：一次会话中包含多次请求和响应。一次会话：浏览器第一次给服务器资源发送请求，会话建立，直到有一方断开为止功能：在一次会话的范围内
- 初步探索首先我们要了解equals方法是什么，hashcode方法是什么。equals方法equals 是java的obejct类的一个方法
- 一个简单的红包生成算法，代码如下：/** * 红包 * @param n * @param money 单位：分 * @return **/
- 详解HDFS多文件Join操作的实例最近在做HDFS文件处理之时，遇到了多文件Join操作，其中包括：All Join以及常用的Left J
