spring-kafka:在Spring Boot中使用交互式查询
spring-kafka是一个用于在Spring Boot应用程序中使用Apache Kafka的库。它提供了一组易于使用的API，使得在Spring Boot应用程序中使用Kafka变得更加简单。
Apache Kafka是一个分布式流处理平台，具有高度可扩展性和持久性。它主要用于构建实时数据管道和流式处理应用程序。Kafka使用了发布-订阅模式，允许多个消费者同时订阅一个或多个主题，并从中读取和写入数据。
使用spring-kafka，可以在Spring Boot应用程序中轻松地创建Kafka生产者和消费者。下面是一些spring-kafka的主要优势和应用场景：
- 优势：
- 集成简单：spring-kafka提供了与Spring Boot框架的无缝集成，使得使用Kafka变得更加简单。
- 高性能：由于Kafka本身具有高吞吐量和低延迟的特性，因此使用spring-kafka可以获得出色的性能。
- 可靠性：Kafka具有分布式的复制机制，可以提供高度可靠的消息传递。
- 弹性扩展：Kafka允许根据需求增加或减少主题的分区数量，以满足不同规模的应用需求。
- 应用场景：
- 实时数据流处理：通过使用Kafka提供的流式处理能力，可以实时处理和分析大规模数据流，例如实时分析日志数据、监控数据等。
- 异步通信：使用Kafka作为消息中间件，可以实现应用程序之间的异步通信，提高系统的可伸缩性和响应性。
- 分布式应用集成：Kafka可以用于将不同的分布式应用程序集成在一起，实现数据共享和协作。
如果你希望在Spring Boot应用程序中使用Kafka，可以使用spring-kafka库。你可以通过以下链接了解更多关于spring-kafka的信息： spring-kafka GitHub
在腾讯云中，你可以使用腾讯云的消息队列CMQ或者物联网通信平台TIoT来实现类似的功能。可以参考以下链接了解腾讯云的相关产品：
相关·内容
在Kafka Streams + Spring Boot应用程序中配置(提供主机和端口信息)和访问交互式查询的惯用方式是什么？ 访问KafkaStreams实例以访问状态存储的正确方法是什么？ 我知道spring-cloud-stream中有InteractiveQueriesService，但我并不是只在Spring Boot中使用spring-kafka库。 谢谢
浏览 11提问于2020-10-17得票数 1
回答已采纳
我们目前正在尝试创建一个SCS应用程序，该应用程序可以生成特定的主题消息，但是在运行该应用程序时，我们遇到了错误： The following method did not exist:
org.springframework.kafka.core.DefaultKafkaProducerFactory.setBeanName(Ljava/lang/String;)V
The method's class, org.springframework.kafka.core.DefaultKafkaProducerFactory, is available from the f
浏览 109提问于2020-07-22得票数 1
回答已采纳
我在spring boot中使用cron表达式安排了一个作业。当我添加maven spring-kafka依赖项时，我的工作没有启动。这是我的pom.xml中的依赖项： <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency> 在我的类中，我使用以下注释
浏览 24提问于2020-09-28得票数 0
我正在使用gradle 4.x，这是我的问题所在。
我有两个模块A和B。
模块B对模块A有项目依赖性。下面是gradle文件。
模块A的build.gradle
apply plugin: 'java-library'
dependencies {
implementation group: 'org.springframework.boot', name: 'spring-boot-starter'
implementation group: 'org.springframework.kafka', name:
浏览 0提问于2019-10-23得票数 0
我有一个Spring Boot (2.3.3)服务，它使用spring-kafka来访问专用的Kafka/Zookeeper配置。我一直在使用服务器设置spring.kafka.bootstrap- application.properties =localhost:9092来访问我的开发/测试Apache Kafka服务。 然而，在生产中，我们在Zookeeper中配置了一个Kafka Brokers集群(在许多服务器上)，并且我被要求修改我的服务来查询Zookeeper，以获得代理列表并使用该列表而不是引导服务器配置。原因是，众所周知，我们的DevOps人员会重新配置服务器/节点和Kaf
浏览 25提问于2020-09-15得票数 1
我的问题可能听起来很奇怪，但仍然。在没有Spring Boot自动配置功能的情况下，是否可以仅将spring-kafka用作配置框架-类似于解析application.properties或application.yaml，并实例化消费者和生产者？基本上，只为public static void main应用程序提供了最少的spring和spring-kafka。
谢谢
浏览 12提问于2020-03-04得票数 0
回答已采纳
我已经将spring boot版本从2.5.3升级到了2.5.5，它已经升级了这个库。 spring-kafka : from 2.7.4 to 2.7.7
spring-kafka-test : from 2.7.4 to 2.7.7
kafka-clients : from 2.7.1 to 2.8.1
spring-test : from 5.3.9 to 5.3.10 在这次升级中，我开始收到一个错误: java.lang.NoSuchMethodError:kafka.server.KafkaServer.(Lkafka/ser
浏览 83提问于2021-10-16得票数 0
回答已采纳
我们有一个spring boot应用程序，它使用spring-kafka。当我们更新一个配置属性时，我们想要禁用kafka生产者(kafkatemplate)。我尝试过使用条件bean和应用程序上下文刷新。
有没有办法用spring-kafka优雅地关闭kafka生产者？
浏览 7提问于2017-02-02得票数 1
回答已采纳
在1.3.*和2.2.*版本之间有没有spring-kafka迁移指南？或者至少一些changelog，这样我就知道我需要查看哪些配置才能进行更改？我在官方文档和项目的github存储库中都找不到任何东西。或者我只需要增加spring-boot和spring-kafka的依赖项，它就可以开箱即用了？
浏览 4提问于2019-03-16得票数 0
回答已采纳
我想在Spring中使用Kafka客户端。
在阅读了之后，
我添加了maven依赖项。
然后我试着启动Spring。它抛出
ClassNotFoundException。这个类是'org.springframework.kafka.listener.RecordInterceptor‘
任何人都能解决这个问题吗？
Version:
<spring.boot.version>2.2.1.RELEASE</spring.boot.version>
<spring.version>5.2.1.RELEA
浏览 12提问于2019-12-03得票数 0
回答已采纳
1回答
我们正在使用spring boot 2.1.7.RELEASE和spring-kafka 2.2.7.RELEASE。现在，我们计划将spring-kafka版本更新到2.4.0.RELEASE并在应用程序启动时获取以下错误
java.lang.ClassNotFoundException: org.springframework.core.log.LogAccessor
你能建议我这里遗漏了什么吗？
浏览 60提问于2020-02-19得票数 0
回答已采纳
我使用的是spring boot 2.5.4和spring-kafka的相关版本。Spring boot向配置的client-id添加一个后缀。例如，如果您在application.properties中设置此配置: spring.kafka.client-id=siavash 您可以看到它使用siavash-0与代理进行通信，这在文档中也有提到：search word "suffix" in this page 有什么方法可以防止spring boot添加这个后缀吗？还是禁用它？
浏览 288提问于2021-09-11得票数 0
我使用的是spring-kafka 2.5.4和spring boot 2.3.2。我有一些带有@StreamListener注解的方法。我想为所有的方法添加一个通用的拦截器。我已经尝试了@EnableKafka和ConcurrentKafkaListenerContainerFactory bean，但它不起作用。
我也不想自己配置ConsumerFactory，而是让spring-kafka来处理。我只想在现有配置的基础上构建，并在其上添加我的RecordInterceptor。
浏览 1提问于2021-02-22得票数 0
回答已采纳
根据spring boot 2.3发行说明，现在支持优雅关机。spring-kafka 2.5.6也支持它吗？
浏览 0提问于2020-10-05得票数 0
一个在Spring-boot 2.2.6.RELEASE和Spring-kafka 2.4.1.RELEASE上运行我们的测试套件的项目在升级到Spring-boot 2.3.1.RELEASE和Spring-kafka 2.5.3.RELEASE时遇到了问题。它似乎找不到kafka.utils.TestUtils。
我已经尝试将spring-kafka版本留空，以便允许spring-boot提取一个兼容的版本，但错误仍然存在。
build.gradle文件：
dependencies {
api "org.springframework.boot:spring-boot-star
浏览 0提问于2020-07-15得票数 0
我有一个应用程序，它动态地生成几个具有相同依赖关系(spring boot，kafka等)的web应用程序(war文件)。.my问题是，我如何将所有这些依赖关系放在Tomcat中以减小war文件的大小，然后在我的pom.xml中编写示例。 <dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<scope>provided</scope&g
浏览 13提问于2020-04-10得票数 0
我现在使用的是spring boot 2和JUnit5。我想使用spring-kafka-test进行测试，但是我使用的spring-kafka版本是2.1.8版本。因此我认为不可能使用EmbeddedKafkaBroker，因为它是2.2.x中的东西。 我使用的kafka客户端是1.1.0。我看了一下https://spring.io/projects/spring-kafka中提到的兼容性矩阵，它似乎不适用于client 1.1.0，只是想知道是否有可能只将spring-kafka-test更新到2.2.3版本，以便在Junit 5中使用EmbeddedKafkaBroker，同时仍然使
浏览 26提问于2019-01-24得票数 0
回答已采纳
3回答
在groovy脚本中，我可以将数组中的依赖项列表指定为"implementation libraries.frameworkLibs"，kotlin中是否存在类似的方法？ 目标是在数组中有一个依赖项列表，这样每个列表都可以在子项目中单独分配。 plugins {
id 'org.springframework.boot' version '2.1.4.RELEASE'
id 'java'
}
apply plugin: 'io.spring.dependency-management'
sou
浏览 15提问于2019-05-08得票数 0
回答已采纳
我正在与Spring-Kafka和Boot合作开发一个项目，希望在application.properties中获得spring.kafka.consumer.client-Id属性的主机名，以便在服务器端日志中区分每个消费者。
有没有办法让我这么做？我查看了spring boot参考指南和java.lang.System类，但没有找到有用的指导。
浏览 11提问于2017-04-04得票数 3
回答已采纳
我使用的是spring boot 2.1.5和spring-kafka，版本未指定。项目是gradle 6.0，java 1.8 我得到了这个错误 error creating bean with name 'org.springframework.boot.autoconfigure.kafka.kafkaannotationdrivenconfiguration': unexpected exception during bean creation; nested exception is java.lang.typenotpresentexception: type o
浏览 102提问于2020-04-25得票数 0
回答已采纳
1回答
如何包含spring kafka的特定版本。Kafka版本依赖是由starter parent导入的2.5.14。但是我想用2.6.10版本覆盖它，我该怎么做呢？ 这是我的Spring Boot starter <parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.12.RELEASE</version>
浏览 47提问于2021-09-28得票数 0
回答已采纳
1回答
我正在开发一个示例Spring集成项目，其中有使用spring集成的kafka侦听器。这在较低版本的boot (boot - 1.5.9，kafka-client 1.0.0，spring-integration-kafka 2.3.0，spring-kafka 1.3.2)上运行良好。但在将启动版本从1.5.9升级到2.1.x时出现以下错误:- Caused by: java.lang.ClassNotFoundException: org.springframework.kafka.listener.config.ContainerProperties
at java.net.U
浏览 106提问于2020-04-28得票数 0
回答已采纳
我使用Spring Boot 2.0.7和Spring-Kafka来创建请求/应答模式。基本上，前端UI向API发出请求，API将消息放到请求Kafka队列中，消息由后端进程处理，完成后将消息放到应答队列中。
我想提供的前端用户界面的api，等待，直到响应准备就绪。此时的UI将仅显示一条正在处理的消息。如果响应不可用(例如，2分钟后)，API应该只返回消息不可用错误，我们可以指示用户稍后再来。
我是Spring-Kafka的新手。它允许我创建一个轮询API吗？如果是这样的话，任何示例代码都将非常感谢。
浏览 1提问于2018-12-08得票数 0
1回答
有没有人知道Spring Boot与Kafka Connect的集成？有一个spring-kafka项目，我认为它与Kafka客户端进行了很好的集成，但没有连接和流API。
浏览 2提问于2017-08-09得票数 7
在spring-boot (版本2.1.4)应用程序中，需要将apache-kafka迁移到spring-kafka。当前的kafka使用者做: 1)在应用程序启动时初始化的apache 2)它有"0“主题分区集3)使用KafkaConsumer kafkaConsumer将数据轮询到ConsumerRecord 4)应用程序自己的重试机制等待并再次轮询直到max_retry 遗留代码如下所示： while (!done.get()) {
ConsumerRecords<byte[], <byte[]> records = kafkaConsumer.po
浏览 21提问于2019-04-15得票数 0
我构造了一个spring boot应用程序来在Kafka中生成数据，但现在我需要将该应用程序连接到旧的Kafka (0.11v)。我尝试只更改bootstrap-server，但收到连接错误。
我再次检查了这个Kafka实例是否工作正常，以及其他应用程序是否正在连接。
我的问题是，像v2.5.4.RELEASE这样的spring-Kafka的新版本是否可以连接到旧版本的Kafka。旧版Kafka实例有什么配置可以添加吗？我构造了一个spring boot应用程序来在Kafka中生成数据，但现在我需要将该应用程序连接到旧的Kafka (0.11v)。我尝试只更改bootstrap-server，
浏览 3提问于2020-08-28得票数 0
当我试图在gradle中构建我的spring引导项目时，我遇到了以下错误。我从弹簧初始化器得到了build.gradle。
无法解析配置的所有工件“：类路径”。找不到spring-boot-gradle-plugin-2.3.0.BUILD-SNAPSHOT.jar (org.springframework.boot:spring-boot-gradle-plugin:2.3.0.BUILD-SNAPSHOT:20200409.145011-519).在以下位置搜索：
请找到我的定位器
plugins {
id 'org.springframework.boot' v
浏览 7提问于2020-04-09得票数 2
我正在写一个使用Kafka Streams，Spring-Kafka和Spring Boot的流应用程序。在使用Spring-Kafka时，我找不到任何信息来正确测试Kafka Streams DSL所做的流处理。文档提到了EmbeddedKafkaBroker，但似乎没有关于如何处理测试的信息，例如状态存储。 只是提供一些简单的例子来说明我想要测试的内容。我注册了以下bean (其中Item是avro生成的)： @Bean
public KTable<String, Long> itemTotalKTable(StreamsBuilder streamsBuil
浏览 25提问于2019-08-31得票数 3
1回答
这是有线刚开始使用简单的main类的spring引导项目，它没有spring依赖，但是在添加了spring-kafka和spring-kafka-test之后异常地崩溃了，github 。
gradle.build
apply plugin: 'java-library'
repositories {
jcenter()
}
dependencies {
compile group: 'org.springframework.boot', name: 'spring-boot-starter-web', version:
浏览 0提问于2018-12-17得票数 6
回答已采纳
我需要识别数据库连接失败异常与其他异常，并在我的kafka侦听器错误处理程序中进行相应的处理。由于Spring Data JPA中的所有内容都被转换为DataAcccessException，我只是想知道当应用程序运行时发生连接故障时会抛出什么异常。工作环境: Spring boot 2.x、Spring data jpa、spring-kafka、Java 8
浏览 2提问于2019-02-20得票数 1
我将spring引导从1.3.5升级到1.5.17，不再找到log4j库。错误表示错误：(10，24) java: package org.apache.log4j不存在。这是我的pom.xml
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.17.RELEASE</version>
<r
浏览 2提问于2019-10-03得票数 0
回答已采纳
我使用的是spring boot 2.1.7.RELEASE和spring-kafka 2.2.8.RELEASE，我们正在升级spring boot版本，但目前我们使用的是spring-kafka版本。 我正在使用@KafkaListener注解来创建一个消费者，并且我正在使用的consumer.And的所有默认设置都是在Spring-Kafka文档中指定的。 // other props
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.clas
浏览 49提问于2021-06-26得票数 0
回答已采纳
我正在用几种不同的技术制作一个Gradle项目，其中之一是Spring合同。
根据这个，生产者在执行:build任务后将存根发布到本地maven存储库。
我正在尝试使用Gradle来执行:build和:publishToMavenLocal任务。它发布所有东西，甚至是普通的jar，而不是存根！
我怎样才能发布存根？
项目的生成文件夹：
项目的本地maven存储库：
如您所见，存根jar没有发布到maven本地。
我的gradle.build
buildscript {
dependencies {
classpath "org.springf
浏览 3提问于2022-07-08得票数 1
回答已采纳
我已经从Confluent安装了Apache Kafka (使用rest-api和schema-registry)。我使用spring-boot和spring-kafka实现了producer (每个事件都在数据库触发器中生成了uuid )。现在我想把它作为多个实例来运行--但是在kafka中省略重复的正确方法是什么呢？如果我在负载均衡器后面有几个spring-boot实例，并且每个实例都有相同生成的uuid的事件，如何只发送一个到kafka？我可以在spring中生成uuid，但是事件的内容仍然是相同的，并且没有必要从每个实例发送它(数据库的触发器无论如何都会将内容发送给每个生产者)。
浏览 0提问于2017-12-13得票数 0
2回答
在我的Spring项目中，我直接返回一个列表。然后我可以在Postman中看到JSON的响应。我没有做任何事情来将Java对象转换为JSON。它怎麽工作?JSON是为Java交换数据的默认方式吗？
这里是一个控制器示例。
@GetMapping("/")
@Secured({"ROLE_ANONYMOUS", "ROLE_REGULAR"})
List<Project> allProducts() {
return projectService.getAllProjects();
}
下面是我的pom.xml的依赖部分
<
浏览 6提问于2022-05-31得票数 0
我正在用springboot 2做项目。 并且想要制作my common library模块。 所以我制作了common和Module A模块。 我想使用mysql-connector-java:5.1.40，所以我在common模块的build.gradle依赖项中进行了设置。 但是当我将common导入到Module A时，Module A有mysql-connector-java:8.0.15 我的build.gradle文件就像这样。 springBootVersion：2.1.4.Relase common的build.gradle依赖 dependencies {
com
浏览 35提问于2019-05-31得票数 0
回答已采纳
我正在尝试基于@KafkaListener注释创建自定义元注释，但是spring-kafka抛出了一个错误。
下面是我的注解：
@Target({ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@KafkaListener
public @interface MessageListener {
@AliasFor(annotation = KafkaListener.class, attribute = &#
浏览 0提问于2017-12-13得票数 1
1回答
我正在使用以下包来消费kafka消息 compile 'org.springframework.boot:spring-boot-starter-webflux'
compile("org.springframework.boot:spring-boot-starter-web")
// tag::actuator[]
compile("org.springframework.boot:spring-boot-starter-actuator")
compile('org.springframework.kafka:spring-kaf
浏览 7提问于2019-05-15得票数 0
回答已采纳
0回答
我在spring-kafka with spring-boot上找到了很多教程，并成功地实现了它们。为什么所有的教程都使用spring-boot？spring-kafka中是否存在某种依赖关系，或者这是最好的解决方案？
我已经有了一个spring项目，并且我正尝试在其中添加一个spring-kafka maven模块。我的主项目的pom.xml如下所示。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
浏览 1提问于2017-06-09得票数 0
回答已采纳
目前我正在使用Spring boot 2.4.0和spring-kafka。 我想使用消费者和生产者，而不是kafka admin。 我试图通过将"bootstrap.servers“设置为null来覆盖KafkaAdmin类，但不起作用。它仍然“尝试”连接以生成KafkaAdmin并记录错误。这不是致命的，但有没有办法完全禁用kafka管理？
浏览 50提问于2021-01-05得票数 0
回答已采纳
我有一个小的库，它提供了一个围绕EmbeddedKafkaBroker的JUnit 5扩展，它试图通过将测试之间的偏移量重置为最新而不是丢弃整个上下文来提高性能，而不仅仅是使用@DirtiesContext。 我的build.gradle是： import org.springframework.boot.gradle.plugin.SpringBootPlugin
plugins {
id 'org.springframework.boot' version '2.3.5.RELEASE' apply false
id 'io.sp
浏览 37提问于2020-11-04得票数 4
回答已采纳
我使用Gradle构建我的spring-boot应用程序。我要什么吗？
我有一个多模块应用程序，其中一些模块是spring-boot应用程序。我希望在父项目的Gradle文件中获得BOM spring依赖项，并使用所有没有版本的库。
现在我有了另一个秘密：
buildscript {
repositories {
mavenCentral()
}
dependencies {
classpath("org.springframework.boot:spring-boot-gradle-pl
浏览 0提问于2019-04-03得票数 1
我使用Spring boot，特别是spring-kafka库来使用kafka。 我在一个使用消息的方法上有一个带有@KafkaListener注释的类。我能够读入消息。我意识到这是一个话题，基于制作人的消息不断涌入。我想要某种提示，我在最后的偏移量上，这样我就可以用另一个source.How触发我的重新调解逻辑，我可以在那里获得这个cue.Is，我可以订阅/收听吗？
浏览 11提问于2020-01-22得票数 0
回答已采纳
1回答
我试图为与卡夫卡一起工作做一个简单的申请。应用程序的目的是侦听kafka主题并对来自它的消息做出反应(在本例中，只是记录消息)。
src/main/java/pkg/Application.java
package pkg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.
浏览 0提问于2018-04-03得票数 0
回答已采纳
测试无法用错误的Cannot resolve method 'parse' in 'DockerImageName'编译
@SpringBootTest
@Testcontainers
public class KafkaContainerTest {
@ClassRule
public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.2.1"));
@Test
publ
浏览 7提问于2020-10-17得票数 1
回答已采纳
我们正在使用spring boot将我们的项目迁移到微服务。我有一个需求，我应该能够使用配置细节运行生产者和消费者，比如引导服务器和用户在运行时提供的键值serdes。
在之前的项目中，我可以使用java Apache Kafka API来实现这一点，但我看不到使用spring-kafka API的任何方法，因为它只允许通过Spring java配置类或application.properties来定义生产者或消费者相关的配置。
浏览 14提问于2021-05-25得票数 1
1回答
春季卡夫卡版本不兼容
、、、、
我试图将kafka集成到项目中的一个服务中，但是它已经集成到了另一个服务中，因此我自然尝试使用这两个依赖项使用相同的kafka版本：
implementation 'org.springframework.integration:spring-integration-kafka:3.1.5.RELEASE'
implementation 'org.springframework.kafka:spring-kafka:2.2.15.RELEASE'
但我得到的只是这个错误：
Error creating bean with name 'kafkaProdu
浏览 39提问于2022-06-22得票数 0
我们将Spring版本从2.1.6升级到2.2.2。在2.1.6中的Prometheus端点中看到的一些属性在2.2.2中不可见。
kafka_consumer_records_consumed_total_records_totalkafka_consumer_records_lag_recordskafka_consumer_fetch_latency_max_secondskafka_consumer_bytes_consumed_total_bytes_total
但是，这些数据正在MBeans中注册。
kafka_consumer_records_consumed_total_rec
浏览 9提问于2019-12-28得票数 3
回答已采纳
在使用库运行Spring应用程序时，我得到了以下错误：
原因: org.springframework.kafka.support.KafkaHeaderMapper :java.lang.ClassNotFoundException
异常将抛出在以下方法主体的第一行：
@ServiceActivator(inputChannel = "producerChannel")
public MessageHandler kafkaMessageHandler() {
KafkaProducerMessageHandler<String, String>
浏览 0提问于2017-12-07得票数 1
回答已采纳
得到：
创建名为“embeddedKafka”的bean时出错
原因: java.lang.ClassNotFoundException: com.codahale.metrics.Reservoir
<kafka.version>2.8.1</kafka.version>
<spring.kafka.version>2.8.1</spring.kafka.version>
springboot版本: 2.6.3 -不能更改
<dependency>
<groupId>org.s
浏览 17提问于2022-07-25得票数 0
回答已采纳
云服务器
ICP备案
实时音视频
云直播
即时通信 IM
腾讯云开发者
