快捷導(dǎo)航

SpringBoot整合Kafka工具類(lèi)的詳細(xì)代碼

更新時(shí)間：2022年09月26日 16:29:47 作者：鍵盤(pán)命

Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，它可以處理消費(fèi)者在網(wǎng)站中的所有動(dòng)作流數(shù)據(jù)，這篇文章主要介紹了SpringBoot整合Kafka工具類(lèi)的代碼詳解，需要的朋友可以參考下

kafka是什么？

Kafka是由Apache軟件基金會(huì)開(kāi)發(fā)的一個(gè)開(kāi)源流處理平臺(tái)，由Scala和Java編寫(xiě)。Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，它可以處理消費(fèi)者在網(wǎng)站中的所有動(dòng)作流數(shù)據(jù)。這種動(dòng)作（網(wǎng)頁(yè)瀏覽，搜索和其他用戶的行動(dòng)）是在現(xiàn)代網(wǎng)絡(luò)上的許多社會(huì)功能的一個(gè)關(guān)鍵因素。這些數(shù)據(jù)通常是由于吞吐量的要求而通過(guò)處理日志和日志聚合來(lái)解決。對(duì)于像Hadoop一樣的日志數(shù)據(jù)和離線分析系統(tǒng)，但又要求實(shí)時(shí)處理的限制，這是一個(gè)可行的解決方案。Kafka的目的是通過(guò)Hadoop的并行加載機(jī)制來(lái)統(tǒng)一線上和離線的消息處理，也是為了通過(guò)集群來(lái)提供實(shí)時(shí)的消息。

應(yīng)用場(chǎng)景

消息系統(tǒng)： Kafka 和傳統(tǒng)的消息系統(tǒng)（也稱(chēng)作消息中間件）都具備系統(tǒng)解耦、冗余存儲(chǔ)、流量削峰、緩沖、異步通信、擴(kuò)展性、可恢復(fù)性等功能。與此同時(shí)，Kafka 還提供了大多數(shù)消息系統(tǒng)難以實(shí)現(xiàn)的消息順序性保障及回溯消費(fèi)的功能。
存儲(chǔ)系統(tǒng)： Kafka 把消息持久化到磁盤(pán)，相比于其他基于內(nèi)存存儲(chǔ)的系統(tǒng)而言，有效地降低了數(shù)據(jù)丟失的風(fēng)險(xiǎn)。也正是得益于 Kafka 的消息持久化功能和多副本機(jī)制，我們可以把 Kafka 作為長(zhǎng)期的數(shù)據(jù)存儲(chǔ)系統(tǒng)來(lái)使用，只需要把對(duì)應(yīng)的數(shù)據(jù)保留策略設(shè)置為“永久”或啟用主題的日志壓縮功能即可。
流式處理平臺(tái)： Kafka 不僅為每個(gè)流行的流式處理框架提供了可靠的數(shù)據(jù)來(lái)源，還提供了一個(gè)完整的流式處理類(lèi)庫(kù)，比如窗口、連接、變換和聚合等各類(lèi)操作。

下面看下SpringBoot整合Kafka工具類(lèi)的詳細(xì)代碼。

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>

工具類(lèi)

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;
    /**
     * 私有靜態(tài)方法，創(chuàng)建Kafka生產(chǎn)者
     * @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表示只要把消息發(fā)送出去就返回成功；1表示只要Leader收到消息就返回成功；all表示所有副本都寫(xiě)入數(shù)據(jù)成功才算成功
        props.put("acks", "all");
        //重試次數(shù)
        props.put("retries", Integer.MAX_VALUE);
        //批處理的字節(jié)數(shù)
        props.put("batch.size", 16384);
        //批處理的延遲時(shí)間，當(dāng)批次數(shù)據(jù)未滿之時(shí)等待的時(shí)間
        props.put("linger.ms", 1);
        //用來(lái)約束KafkaProducer能夠使用的內(nèi)存緩沖的大小的，默認(rèn)值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);
    }

    /**
     * 私有靜態(tài)方法，創(chuàng)建Kafka消費(fèi)者
     * @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");
        //每個(gè)消費(fèi)者分配獨(dú)立的消費(fèi)者組編號(hào)
        props.put("group.id", "111");
        //如果value合法，則自動(dòng)提交偏移量
        props.put("enable.auto.commit", "true");
        //設(shè)置多久一次更新被消費(fèi)消息的偏移量
        props.put("auto.commit.interval.ms", "1000");
        //設(shè)置會(huì)話響應(yīng)的時(shí)間，超過(guò)這個(gè)時(shí)間kafka可以選擇放棄消費(fèi)或者消費(fèi)下一條消息
        props.put("session.timeout.ms", "30000");
        //自動(dòng)重置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);
    }
    /**
     * 私有靜態(tài)方法，創(chuàng)建Kafka集群管理員對(duì)象
     * @author o
     */
    public static void createAdmin(String servers){
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,servers);
        admin = AdminClient.create(props);
    }

    /**
     * 私有靜態(tài)方法，創(chuàng)建Kafka集群管理員對(duì)象
     * @author o
     * @return AdminClient
     */
    private static void createAdmin(){
        createAdmin("node01:9092,node02:9092,node03:9092");
    }

    /**
     * 傳入kafka約定的topic,json格式字符串，發(fā)送給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消費(fèi)數(shù)據(jù)，用于測(cè)試，數(shù)據(jù)最終會(huì)輸出到控制臺(tái)上
     * @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數(shù)組，消費(fèi)數(shù)據(jù)
     * @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格式字符串?dāng)?shù)組，發(fā)送給kafka集群
     * 用于批量發(fā)送消息，性能較高。
     * @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集合，內(nèi)部轉(zhuǎn)為json發(fā)送給kafka集群 <br>
     * 用于批量發(fā)送消息，性能較高。
     * @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，內(nèi)部轉(zhuǎn)為json發(fā)送給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();
    }

    /**
     * 創(chuàng)建主題
     * @author o
     * @param name 主題的名稱(chēng)
     * @param numPartitions 主題的分區(qū)數(shù)
     * @param replicationFactor 主題的每個(gè)分區(qū)的副本因子
     */
    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)));
        //以下內(nèi)容用于判斷創(chuàng)建主題的結(jié)果
        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 主題的名稱(chēng)
     */
    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);
        //以下內(nèi)容用于判斷刪除主題的結(jié)果
        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 主題的名稱(chēng)
     */
    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);
        //以下內(nèi)容用于顯示主題詳情的結(jié)果
        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());
    }
}

到此這篇關(guān)于SpringBoot整合Kafka工具類(lèi)的文章就介紹到這了,更多相關(guān)SpringBoot整合Kafka工具類(lèi)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: