溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇內容介紹了“SpringBoot怎么整合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工具類”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
