spring boot 使用 Kafka的场景分析

更新时间：2025年12月30日 09:11:13 作者：奋力向前123

本文详细介绍了Kafka作为消息队列在SpringBoot中的使用方法,包括添加依赖、创建生产者和消费者,以及与RocketMQ的比较,着重于数据可靠性、性能和消息传递方式,还探讨了Kafka在实时数据流处理、事件驱动架构等场景的应用,感兴趣的朋友跟随小编一起看看吧

一、Kafka作为消息队列的好处

高吞吐量：Kafka能够处理大规模的数据流，并支持高吞吐量的消息传输。
持久性：Kafka将消息持久化到磁盘上，保证了消息不会因为系统故障而丢失。
分布式：Kafka是一个分布式系统，可以在多个节点上运行，具有良好的可扩展性和容错性。
支持多种协议：Kafka支持多种协议，如TCP、HTTP、UDP等，可以与不同的系统进行集成。
灵活的消费模式：Kafka支持多种消费模式，如拉取和推送，可以根据需要选择合适的消费模式。
可配置性强：Kafka的配置参数非常丰富，可以根据需要进行灵活配置。
社区支持：Kafka作为Apache旗下的开源项目，拥有庞大的用户基础和活跃的社区支持，方便用户得到及时的技术支持。

二、springboot中使用Kafka

添加依赖：在pom.xml文件中添加Kafka的依赖，包括spring-kafka和kafka-clients。确保版本与你的项目兼容。
创建生产者：创建一个Kafka生产者类，实现Producer接口，并使用KafkaTemplate发送消息。
配置生产者：在Spring Boot的配置文件中配置Kafka生产者的相关参数，例如bootstrap服务器地址、Kafka主题等。
发送消息：在需要发送消息的地方，注入Kafka生产者，并使用其发送消息到指定的Kafka主题。
创建消费者：创建一个Kafka消费者类，实现Consumer接口，并使用KafkaTemplate订阅指定的Kafka主题。
配置消费者：在Spring Boot的配置文件中配置Kafka消费者的相关参数，例如group id、auto offset reset等。
接收消息：在需要接收消息的地方，注入Kafka消费者，并使用其接收消息。
处理消息：对接收到的消息进行处理，例如保存到数据库或进行其他业务逻辑处理。

三、使用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>

创建生产者：创建一个Kafka生产者类，实现Producer接口，并使用KafkaTemplate发送消息。

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都是开源的消息队列系统，它们具有许多相似之处，但在一些关键方面也存在差异。以下是它们在数据可靠性、性能、消息传递方式等方面的比较：

数据可靠性：

Kafka使用异步刷盘方式，而RocketMQ支持异步实时刷盘、同步刷盘、同步复制和异步复制。这使得RocketMQ在单机可靠性上比Kafka更高，因为它不会因为操作系统崩溃而导致数据丢失。此外，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发送消息后，broker马上把消息投递给consumer，这种方式实时性较高，但会增加broker的负载。而RocketMQ基于Pull模式和Push模式的长轮询机制，来平衡Push和Pull模式各自的优缺点。RocketMQ的消息及时性较好，严格的消息顺序得到了保证。

其他特性：

Kafka在单机支持的队列数超过64个队列，而RocketMQ最高支持5万个队列。队列越多，可以支持的业务就越多。

五、kafka使用场景

实时数据流处理：Kafka可以处理大量的实时数据流，这些数据流可以来自不同的源，如用户行为、传感器数据、日志文件等。通过Kafka，可以将这些数据流进行实时的处理和分析，例如进行实时数据分析和告警。
消息队列：Kafka可以作为一个消息队列使用，用于在分布式系统中传递消息。它能够处理高吞吐量的消息，并保证消息的有序性和可靠性。
事件驱动架构：Kafka可以作为事件驱动架构的核心组件，将事件数据发布到不同的消费者，以便进行实时处理。这种架构可以简化应用程序的设计和开发，提高系统的可扩展性和灵活性。
数据管道：Kafka可以用于数据管道，将数据从一个系统传输到另一个系统。例如，可以将数据从数据库或日志文件传输到大数据平台或数据仓库。
业务事件通知：Kafka可以用于通知业务事件，例如订单状态变化、库存更新等。通过订阅Kafka主题，相关的应用程序和服务可以实时地接收到这些事件通知，并进行相应的处理。
流数据处理框架集成：Kafka可以与流处理框架集成，如Apache Flink、Apache Spark等。通过集成，可以将流数据从Kafka中实时导入到流处理框架中进行处理，实现流式计算和实时分析。

到此这篇关于spring boot 使用 Kafka的场景分析的文章就介绍到这了,更多相关spring boot 使用 Kafka内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

java super关键字知识点详解
在本篇文章里小编给大家整理的是一篇关于java super关键字知识点详解内容，有兴趣的朋友们可以参考下。
2021-01-01
Spring将MultipartFile转存到本地磁盘的三种方式
在Java中处理文件向来是一种不是很方便的操作,然后随着Spring框架的崛起,使用Spring框架中的MultipartFile来处理文件也是件很方便的事了,今天就给大家介绍Spring将MultipartFile转存到本地磁盘的方式,需要的朋友可以参考下
2024-10-10
Java实现更新顺序表中的指定元素的示例
本文主要介绍了Java实现更新顺序表中的指定元素的示例，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-06-06
SpringBoot结合mockito测试实战
与集成测试将系统作为一个整体测试不同，单元测试更应该专注于某个类。所以当被测试类与外部类有依赖的时候，尤其是与数据库相关的这种费时且有状态的类，很难做单元测试。但好在可以通过“Mockito”这种仿真框架来模拟这些比较费时的类，从而专注于测试某个类内部的逻辑
2022-11-11
简单介绍Java编程中的线程池
这篇文章主要介绍了Java编程中的线程池,进程和线程的并发是Java编程中的重要环节,需要的朋友可以参考下
2015-09-09
详解maven的setting配置文件中mirror和repository的区别
这篇文章主要介绍了详解maven的setting配置文件中mirror和repository的区别，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2018-12-12
SpringBoot将多个文件夹进行压缩的两种方法(浏览器下载和另存为)
Spring Boot项目通常不会自动对文件夹进行压缩,不过,在打包应用时,如果你使用了Maven或Gradle这样的构建工具,并且配置了相应的插件,可以在打成jar或war包的时候将依赖的库文件合并并压缩,本文介绍了SpringBoot将多个文件夹进行压缩的两种方法
2024-07-07
关于Arrays.sort()使用的注意事项
这篇文章主要介绍了关于Arrays.sort()使用的注意事项，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-05-05
本地安装MinIO分布式对象存储服务器的详细步骤
本地安装MinIO非常简单,MinIO提供了独立的二进制文件,无需额外的依赖,本文介绍如何在本地安装MinIO分布式对象存储服务器,感兴趣的朋友一起看看吧
2024-01-01
Spring AI TikaDocumentReader详解
TikaDocumentReader是SpringAI中用于从多种格式文档中提取文本内容的组件,支持PDF、DOC/DOCX、PPT/PPTX和HTML等格式,它在构建知识库、文档处理和数据清洗等任务中非常有用
2025-01-01