spring boot 使用 Kafka的場(chǎng)景分析

更新時(shí)間：2025年12月30日 09:11:13 作者：奮力向前123

本文詳細(xì)介紹了Kafka作為消息隊(duì)列在SpringBoot中的使用方法,包括添加依賴、創(chuàng)建生產(chǎn)者和消費(fèi)者,以及與RocketMQ的比較,著重于數(shù)據(jù)可靠性、性能和消息傳遞方式,還探討了Kafka在實(shí)時(shí)數(shù)據(jù)流處理、事件驅(qū)動(dòng)架構(gòu)等場(chǎng)景的應(yīng)用,感興趣的朋友跟隨小編一起看看吧

一、Kafka作為消息隊(duì)列的好處

高吞吐量：Kafka能夠處理大規(guī)模的數(shù)據(jù)流，并支持高吞吐量的消息傳輸。
持久性：Kafka將消息持久化到磁盤上，保證了消息不會(huì)因?yàn)橄到y(tǒng)故障而丟失。
分布式：Kafka是一個(gè)分布式系統(tǒng)，可以在多個(gè)節(jié)點(diǎn)上運(yùn)行，具有良好的可擴(kuò)展性和容錯(cuò)性。
支持多種協(xié)議：Kafka支持多種協(xié)議，如TCP、HTTP、UDP等，可以與不同的系統(tǒng)進(jìn)行集成。
靈活的消費(fèi)模式：Kafka支持多種消費(fèi)模式，如拉取和推送，可以根據(jù)需要選擇合適的消費(fèi)模式。
可配置性強(qiáng)：Kafka的配置參數(shù)非常豐富，可以根據(jù)需要進(jìn)行靈活配置。
社區(qū)支持：Kafka作為Apache旗下的開源項(xiàng)目，擁有龐大的用戶基礎(chǔ)和活躍的社區(qū)支持，方便用戶得到及時(shí)的技術(shù)支持。

二、springboot中使用Kafka

添加依賴：在pom.xml文件中添加Kafka的依賴，包括spring-kafka和kafka-clients。確保版本與你的項(xiàng)目兼容。
創(chuàng)建生產(chǎn)者：創(chuàng)建一個(gè)Kafka生產(chǎn)者類，實(shí)現(xiàn)Producer接口，并使用KafkaTemplate發(fā)送消息。
配置生產(chǎn)者：在Spring Boot的配置文件中配置Kafka生產(chǎn)者的相關(guān)參數(shù)，例如bootstrap服務(wù)器地址、Kafka主題等。
發(fā)送消息：在需要發(fā)送消息的地方，注入Kafka生產(chǎn)者，并使用其發(fā)送消息到指定的Kafka主題。
創(chuàng)建消費(fèi)者：創(chuàng)建一個(gè)Kafka消費(fèi)者類，實(shí)現(xiàn)Consumer接口，并使用KafkaTemplate訂閱指定的Kafka主題。
配置消費(fèi)者：在Spring Boot的配置文件中配置Kafka消費(fèi)者的相關(guān)參數(shù)，例如group id、auto offset reset等。
接收消息：在需要接收消息的地方，注入Kafka消費(fèi)者，并使用其接收消息。
處理消息：對(duì)接收到的消息進(jìn)行處理，例如保存到數(shù)據(jù)庫或進(jìn)行其他業(yè)務(wù)邏輯處理。

三、使用Kafka

pom中填了依賴

<dependency>  
    <groupId>org.springframework.kafka</groupId>  
    <artifactId>spring-kafka</artifactId>  
    <version>2.8.1</version>  
</dependency>  
<dependency>  
    <groupId>org.apache.kafka</groupId>  
    <artifactId>kafka-clients</artifactId>  
    <version>2.8.1</version>  
</dependency>

創(chuàng)建生產(chǎn)者：創(chuàng)建一個(gè)Kafka生產(chǎn)者類，實(shí)現(xiàn)Producer接口，并使用KafkaTemplate發(fā)送消息。

import org.apache.kafka.clients.producer.*;  
import org.springframework.beans.factory.annotation.Value;  
import org.springframework.kafka.core.KafkaTemplate;  
import org.springframework.stereotype.Component;  
@Component  
public class KafkaProducer {  
    @Value("${kafka.bootstrap}")  
    private String bootstrapServers;  
    @Value("${kafka.topic}")  
    private String topic;  
    private KafkaTemplate<String, String> kafkaTemplate;  
    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {  
        this.kafkaTemplate = kafkaTemplate;  
    }  
    public void sendMessage(String message) {  
        Producer<String, String> producer = new KafkaProducer<>(bootstrapServers, new StringSerializer(), new StringSerializer());  
        try {  
            producer.send(new ProducerRecord<>(topic, message));  
        } catch (Exception e) {  
            e.printStackTrace();  
        } finally {  
            producer.close();  
        }  
    }  
}

配置生產(chǎn)者：在Spring Boot的配置文件中配置Kafka生產(chǎn)者的相關(guān)參數(shù)，例如bootstrap服務(wù)器地址、Kafka主題等。

import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  
import org.springframework.kafka.core.DefaultKafkaProducerFactory;  
import org.springframework.kafka.core.KafkaTemplate;  
import org.springframework.kafka.core.ProducerFactory;  
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;  
import org.springframework.kafka.core.ConsumerFactory;  
import org.springframework.kafka.core.ConsumerConfig;  
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;  
import org.springframework.kafka.listener.MessageListener;  
import org.springframework.context.annotation.PropertySource;  
import java.util.*;  
import org.springframework.beans.factory.*;  
import org.springframework.*;  
import org.springframework.*;expression.*;value; 																																		 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	  @Value("${kafka}")   Properties kafkaProps = new Properties(); @Bean public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf){ KafkaTemplate<String, String> template = new KafkaTemplate<>(pf); template .setMessageConverter(new StringJsonMessageConverter()); template .setSendTimeout(Duration .ofSeconds(30)); return template ; } @Bean public ProducerFactory<String, String> producerFactory(){ DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(kafkaProps); factory .setBootstrapServers(bootstrapServers); factory .setKeySerializer(new StringSerializer()); factory .setValueSerializer(new StringSerializer()); return factory ; } @Bean public ConsumerFactory<String, String> consumerFactory(){ DefaultKafkaConsumerFactory<String, String> factory = new DefaultKafkaConsumerFactory<>(consumerConfigProps); factory .setBootstrapServers(bootstrapServers); factory .setKeyDeserializer(new StringDeserializer()); factory .setValueDeserializer(new StringDeserializer()); return factory ; } @Bean public ConcurrentMessageListenerContainer<String, String> container(ConsumerFactory<String, String> consumerFactory, MessageListener listener){ ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(consumerFactory); container .setMessageListener(listener); container .setConcurrency(3); return container ; } @Bean public MessageListener

消費(fèi)者

import org.apache.kafka.clients.consumer.*;  
import org.springframework.kafka.core.KafkaTemplate;  
import org.springframework.stereotype.Component;  
@Component  
public class KafkaConsumer {  
    @Value("${kafka.bootstrap}")  
    private String bootstrapServers;  
    @Value("${kafka.group}")  
    private String groupId;  
    @Value("${kafka.topic}")  
    private String topic;  
    private KafkaTemplate<String, String> kafkaTemplate;  
    public KafkaConsumer(KafkaTemplate<String, String> kafkaTemplate) {  
        this.kafkaTemplate = kafkaTemplate;  
    }  
    public void consume() {  
        Consumer<String, String> consumer = new KafkaConsumer<>(consumerConfigs());  
        consumer.subscribe(Collections.singletonList(topic));  
        while (true) {  
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));  
            for (ConsumerRecord<String, String> record : records) {  
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());  
            }  
        }  
    }  
    private Properties consumerConfigs() {  
        Properties props = new Properties();  
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);  
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);  
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");  
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");  
        return props;  
    }  
}

四、kafka與rocketMQ比較

Kafka和RocketMQ都是開源的消息隊(duì)列系統(tǒng)，它們具有許多相似之處，但在一些關(guān)鍵方面也存在差異。以下是它們?cè)跀?shù)據(jù)可靠性、性能、消息傳遞方式等方面的比較：

數(shù)據(jù)可靠性：

Kafka使用異步刷盤方式，而RocketMQ支持異步實(shí)時(shí)刷盤、同步刷盤、同步復(fù)制和異步復(fù)制。這使得RocketMQ在單機(jī)可靠性上比Kafka更高，因?yàn)樗粫?huì)因?yàn)椴僮飨到y(tǒng)崩潰而導(dǎo)致數(shù)據(jù)丟失。此外，RocketMQ新增的同步刷盤機(jī)制也進(jìn)一步保證了數(shù)據(jù)的可靠性。

性能：

Kafka和RocketMQ在性能方面各有千秋。由于Kafka的數(shù)據(jù)以partition為單位，一個(gè)Kafka實(shí)例上可能有多達(dá)上百個(gè)partition，而一個(gè)RocketMQ實(shí)例上只有一個(gè)partition。這使得RocketMQ可以充分利用IO組的commit機(jī)制，批量傳輸數(shù)據(jù)，從而在replication時(shí)具有更好的性能。然而，Kafka的異步replication性能理論上低于RocketMQ的replication，因?yàn)橥絩eplication與異步replication相比，性能上會(huì)有約20%-30%的損耗。

消息傳遞方式：

Kafka和RocketMQ在消息傳遞方式上也有所不同。Kafka采用Producer發(fā)送消息后，broker馬上把消息投遞給consumer，這種方式實(shí)時(shí)性較高，但會(huì)增加broker的負(fù)載。而RocketMQ基于Pull模式和Push模式的長(zhǎng)輪詢機(jī)制，來平衡Push和Pull模式各自的優(yōu)缺點(diǎn)。RocketMQ的消息及時(shí)性較好，嚴(yán)格的消息順序得到了保證。

其他特性：

Kafka在單機(jī)支持的隊(duì)列數(shù)超過64個(gè)隊(duì)列，而RocketMQ最高支持5萬個(gè)隊(duì)列。隊(duì)列越多，可以支持的業(yè)務(wù)就越多。

五、kafka使用場(chǎng)景

實(shí)時(shí)數(shù)據(jù)流處理：Kafka可以處理大量的實(shí)時(shí)數(shù)據(jù)流，這些數(shù)據(jù)流可以來自不同的源，如用戶行為、傳感器數(shù)據(jù)、日志文件等。通過Kafka，可以將這些數(shù)據(jù)流進(jìn)行實(shí)時(shí)的處理和分析，例如進(jìn)行實(shí)時(shí)數(shù)據(jù)分析和告警。
消息隊(duì)列：Kafka可以作為一個(gè)消息隊(duì)列使用，用于在分布式系統(tǒng)中傳遞消息。它能夠處理高吞吐量的消息，并保證消息的有序性和可靠性。
事件驅(qū)動(dòng)架構(gòu)：Kafka可以作為事件驅(qū)動(dòng)架構(gòu)的核心組件，將事件數(shù)據(jù)發(fā)布到不同的消費(fèi)者，以便進(jìn)行實(shí)時(shí)處理。這種架構(gòu)可以簡(jiǎn)化應(yīng)用程序的設(shè)計(jì)和開發(fā)，提高系統(tǒng)的可擴(kuò)展性和靈活性。
數(shù)據(jù)管道：Kafka可以用于數(shù)據(jù)管道，將數(shù)據(jù)從一個(gè)系統(tǒng)傳輸?shù)搅硪粋€(gè)系統(tǒng)。例如，可以將數(shù)據(jù)從數(shù)據(jù)庫或日志文件傳輸?shù)酱髷?shù)據(jù)平臺(tái)或數(shù)據(jù)倉庫。
業(yè)務(wù)事件通知：Kafka可以用于通知業(yè)務(wù)事件，例如訂單狀態(tài)變化、庫存更新等。通過訂閱Kafka主題，相關(guān)的應(yīng)用程序和服務(wù)可以實(shí)時(shí)地接收到這些事件通知，并進(jìn)行相應(yīng)的處理。
流數(shù)據(jù)處理框架集成：Kafka可以與流處理框架集成，如Apache Flink、Apache Spark等。通過集成，可以將流數(shù)據(jù)從Kafka中實(shí)時(shí)導(dǎo)入到流處理框架中進(jìn)行處理，實(shí)現(xiàn)流式計(jì)算和實(shí)時(shí)分析。

到此這篇關(guān)于spring boot 使用 Kafka的場(chǎng)景分析的文章就介紹到這了,更多相關(guān)spring boot 使用 Kafka內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: