spring boot 使用 Kafka的場景分析

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

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

一、Kafka作為消息隊列的好處

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

二、springboot中使用Kafka

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

三、使用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)建生產者：創(chuàng)建一個Kafka生產者類，實現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();  
        }  
    }  
}

配置生產者：在Spring Boot的配置文件中配置Kafka生產者的相關參數，例如bootstrap服務器地址、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

消費者

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都是開源的消息隊列系統(tǒng)，它們具有許多相似之處，但在一些關鍵方面也存在差異。以下是它們在數據可靠性、性能、消息傳遞方式等方面的比較：

數據可靠性：

Kafka使用異步刷盤方式，而RocketMQ支持異步實時刷盤、同步刷盤、同步復制和異步復制。這使得RocketMQ在單機可靠性上比Kafka更高，因為它不會因為操作系統(tǒng)崩潰而導致數據丟失。此外，RocketMQ新增的同步刷盤機制也進一步保證了數據的可靠性。

性能：

Kafka和RocketMQ在性能方面各有千秋。由于Kafka的數據以partition為單位，一個Kafka實例上可能有多達上百個partition，而一個RocketMQ實例上只有一個partition。這使得RocketMQ可以充分利用IO組的commit機制，批量傳輸數據，從而在replication時具有更好的性能。然而，Kafka的異步replication性能理論上低于RocketMQ的replication，因為同步replication與異步replication相比，性能上會有約20%-30%的損耗。

消息傳遞方式：

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

其他特性：

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

五、kafka使用場景

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

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

您可能感興趣的文章: