我是新卡夫卡，弹力靴，并试图整合卡夫卡和弹性搜索在我的弹力启动应用程序。
当我尝试运行springboot应用程序时，我会看到下面的错误：
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'kafkaListenerContainerFactory' defined in class path resource [org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfi
浏览 14提问于2019-10-04得票数 3
我是Spring和Kafka的新手。我正在做一个使用SpringBoot-kafka的用例，允许用户在运行时创建kafka主题。spring应用程序期望在运行时以编程方式订阅这些主题。到目前为止，我所知道的是，Kafka listener是设计时，因此在启动之前需要指定主题。有没有办法在SpringBoot-Kafka集成中动态订阅kafka主题？
已引用此
我计划实现的当前方法是，不要使用Spring-Kafka集成，而是自己使用这里提到的代码实现Kafka消费者
浏览 0提问于2017-01-08得票数 11
尝试使用Springboot-Kafka进行实践。有几个疑问。这些参数对于application.yml中的Kafka调优有效吗 fetch-max-bytes: 52428800 你知道这是不是Kakfa上的有效属性？我在我的springboot项目中的application.yml中使用它来控制我的消费者轮询行为。其次，在此块中是否存在由KafkaListener内部启动的多个线程 @KafkaListener(id = "instance1", groupId = "${spring.kafka.consumer.group-id}", topics =
浏览 55提问于2021-08-05得票数 0
回答已采纳
在其中一个例子中，当消费者服务重新启动时，它会导致重新处理发送给Kafka的所有记录。
Kafka Broker: 0.10.0.1
Kafka producer Service: Springboot version 1.4.3.Release
Kafka Consumer Springboot Service: Springboot version 2.2.0.Release
现在，为了研究这个问题，我想在开发/本地环境中重新创建这个场景，这个场景不会发生！
可能的原因是什么？
如何检查，当我们发送Acknowledgement.acknowledge();时，是否从客户端处理了记录
消费
浏览 3提问于2020-07-10得票数 0
我的系统管道是"IBM队列-> Springboot app (消息格式化程序) --> Kafka topic-1 (消息队列) --> Springboot app (消息转换器)-> --> Kafka Top-2(消息队列)“。
我的要求是测试IBM中生成的消息的端到端处理时间，并通过管道进行处理。我有一个带有JMeter的JMeter负载测试脚本，它生成通过管道泵出的大量消息。目前还没有实现Dynatrace监控，但是这里使用了用于Kafka监控的汇合控制中心。
您能否建议一些可能的选项来度量Springboot应用程序处理的消息的个人和端到端处
浏览 5提问于2019-11-26得票数 0
回答已采纳
使用Spring JDBC模板，您可以使用一个简单的标志来懒惰地初始化连接。对于Springboot 1.5.x / Spring Kafka 1.3.x部署的Kafka容器工厂，是否有类似的功能？
到目前为止，我看到的最好的答案是禁用autoStart并自己管理启动，在这里捕获启动过程中可能发生的任何异常-
这是唯一的方法吗?在使用KafkaListenerEndpointRegistry自我管理容器的生命周期时，是否有任何注意事项？
Lazy注解是否适用于@KafkaListener，@Configuration类的Kafka配置或类似的组件类？提出这个问题，因为似乎没有一个书面的方法来
浏览 0提问于2019-10-29得票数 1
我想使用kafka和springboot实时地从文件中发送数据，我已经测试了这个代码，用postman以JSON格式将数据发送到kafka主题。
@RestController
public class OrderController {
@Autowired
private KafkaProducerService kafkaProducerService;
@PostMapping("/post")
public void order(@RequestBody Order order) {
kafkaProduce
浏览 7提问于2022-11-07得票数 0
我目前正在做一个小项目，我需要将kafka集群连接到mongodb，以便将发布到kafka主题的消息存储在mongodb数据库中。我目前有一个本地kafka集群，一个springboot生产者发布消息到一个主题，一个springboot消费者在本地消费这些消息。我也在本地安装了mongodb compass…我一直在看教程，我想我必须使用某种类型的接收器连接器( mongodb连接器)来做我想做的事情，但大多数示例都是基于mongodb atlas和confluent或亚马逊网络服务托管的kafka...is，有任何好的教程我可以学习如何在本地kafka集群和本地mongodb上做，或者如果你
浏览 1提问于2021-05-09得票数 0
尝试将使用springboot的springboot应用程序从2.8.4升级到2.9.3版本。
我遵循这里的说明，将kafka版本3.2.3作为一个单独的依赖项。
我的依赖关系如下所示
implementation("org.springframework.kafka:spring-kafka:2.9.3") {
exclude group: "org.apache.kafka", module: "kafka-streams"
exclude group: "org.apache.kafka", module:
浏览 22提问于2022-11-30得票数 0
回答已采纳
我在我的项目中使用了Kafka和Springboot，现在我有了外部Kafka集群和SpringBoot微服务项目。
使用yml文件，我成功地创建了Kafka生产者和消费者，然后应用程序和kafka就可以很好地通信了。
现在我想在我的sprintboot项目中使用嵌入式kafka。类似于嵌入在Activemq代码之后。
spring:
# Embedded ActiveMQ Configuration Example
activemq:
broker-url: vm://embedded?broker.persistent=false,useShutdownHook=fa
浏览 5提问于2019-01-18得票数 1
我在一个springboot应用程序中有一个简单的kafka消费者侦听器，如下所示：
@KafkaListener(topics="mytopic")
public void receive(String message) {
LOGGER.info("received message='{}'", messge);
}
在某些特定情况下，我想拒绝该消息，但我希望系统在特定时间后再次向我提出该消息；
我该怎么做呢？
注意:我也希望kafka的配置是定制的(而不是默认的springboot结构)
浏览 11提问于2019-07-11得票数 1
回答已采纳
将数据从mysql传递到kafka produce的最佳方式是什么 我知道有很好的kafka连接器。然而，我想使用springboot框架。因为我需要转换数据并与其他http结果合并，然后我想产生数据到kafka。 那么有没有什么好的例子，github，博客wahtever~？
浏览 44提问于2020-07-21得票数 1
我试图在Spring kafa中实现错误处理。在我的Kafka侦听器中，我抛出一个运行时异常，如下所示：
@KafkaListener(topics= "Kafka-springboot-example", groupId="group-employee-json")
public void consumeEmployeeJson(Employee employee) {
logger.info("Consumed Employee JSON: "+ employee);
if(null==employe
浏览 1提问于2020-04-05得票数 1
回答已采纳
我目前正在创建一个Springboot应用程序，它将使用Kafka主题的消息，处理后将写回另一个主题。我用骆驼来整合。
My Route looks like this :
onException(IllegalArgumentException.class).maximumRedeliveries(4);
from("kafka:CDC-GEXPUAT-CUSTOMER")
.id("CamelRouteCustomer_1")
.**transacted**()
浏览 3提问于2021-05-17得票数 1
1回答
Java 8，
Flink 1.9.1，
Azure事件中心
我不能再连接到蔚蓝事件中心与我的flink项目，从2020年1月5日。我在几个spring引导应用程序中也有同样的问题，但是当我升级到SpringBoot2.2.2时，这个问题得到了解决，SpringBoot2.2.2也将Kafka客户端和Kafka依赖项更新为2.3.1。我试图更新Flink的kafka依赖，但没有成功。我还提交了一个问题
2020-01-10 19:36:30,364 org.apache.kafka.clients.NetworkClient -消费客户端=消费者-1，groupId=**** Bootst
浏览 0提问于2020-01-10得票数 1
回答已采纳
3回答
由于Kafka有一个基于主题的发布子体系结构，如何使用Kafka处理web应用程序的一对一和组消息部分？我正在使用SpringBoot+Angular堆栈和Docker服务器。
浏览 7提问于2020-05-09得票数 6
回答已采纳
我有一个应用程序，它使用2.10.0版本中的SpringBoot 2.10.0 version和kafka。这个应用程序有一个简单的生产者和消费者:发送者使用KafkaTemplate，消费者使用KafkaListener。我试图实现的目标是即使KafkaServer没有运行，也能够启动SpringBoot应用程序。目前，如果没有正在运行的KafkaBroker，应用程序将无法启动，并显示以下错误消息： org.springframework.context.ApplicationContextException:
Failed to start bean 'org.sprin
浏览 216提问于2019-01-25得票数 2
1回答
我用Spring模块创建了一个卡夫卡听众。当我使用JMX打倒所有Kafka侦听器时，Springboot应用程序就会崩溃。在这个应用程序中，除了Kafka使用者之外，我没有其他线程在运行。
浏览 11提问于2021-02-23得票数 0
回答已采纳
我有以下定义：
@KafkaListener(
topics = "${app.kafka.eventTopic}",
groupId = "${app.kafka.consumerGroupId}",
concurrency = "${app.kafka.concurrency}"
)
我在yaml文件中定义了以下配置属性：
app:
kafka:
eventTopic: topicName
consumerGroupId: groupN
浏览 2提问于2021-12-17得票数 0
回答已采纳
当我在本地运行时，我可以看到正在收集kafka.consumer.。当我部署我的服务时，我发现这些指标并不存在。
我使用了kafka版本的1.11.0、java 11和SpringBoot2.2。
我怎么才能确定缺了什么？
浏览 3提问于2019-10-30得票数 0
回答已采纳
2回答
绑定中的输出绑定
、、
在这个中，你不能只做一个输出
但是我只需要用做一个输出
我该怎么办？
一些文章说，使用org.springframework.cloud.stream.function.StreamBridge，但它不适用于我
我使用StreamBridge将主题发送给Kafka，但是Kafka没有为我的Spring应用程序生成主题
这是我的application.yml并生成主题代码
// producer Springboot application
spring.cloud.stream:
kafka:
binder:
brokers: {A
浏览 4提问于2021-08-02得票数 1
回答已采纳
我有一个简单的springboot引导应用程序运行在码头容器与以下配置。
@Configuration
public class ProducerConfiguration {
private static final String KAFKA_BROKER = "kafka:9092";
@Bean
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(produc
浏览 1提问于2020-03-07得票数 1
回答已采纳
我有以下配置：
春季-整合-kafka 1.3.1.RELEASE
我有一个定制的卡夫卡水槽和一个定制的卡夫卡源
我想要的配置：
我仍然想使用Spring kafka 1.3.1.RELEASE和我的定制kafka-接收器。
我正在改变我的kafka源代码逻辑，以使用Spring-integration-kafka -2.1.0 producer。我注意到实现消费者/生产者的方式与以前版本的Spring-integration-kafka大不相同。
我的问题是:我能面对一些相容的问题吗？
浏览 2提问于2017-04-04得票数 1
1回答
我有一个JAVA springboot项目，在kafka中推送avro消息。在我的开发过程中，使用windows PC，我所有的kafka消息都以所有特殊字符(é，à，...)显示。
使用以下序列化程序：
key-serializer: "org.apache.kafka.common.serialization.StringSerializer"
value-serializer: "io.confluent.kafka.serializers.KafkaAvroSerializer"
在使用unix在artifactory上编译和部署我的JAR之后，我的消
浏览 41提问于2021-05-19得票数 0
回答已采纳
我的SpringBoot应用程序正在本地主机上运行，但我希望它与远程服务器下载的Kafka连接，例如123.45.6.789。我能够从mobaxTerm终端生成和使用消息，但由于这个错误，我的springboot应用程序无法与kafka连接。
WARN[0;39m [35m18788[0;39m [2m---[0;39m [2m[ntainer#0-0-C-1][0;39m
[36morg.apache.kafka.clients.NetworkClient [0;39m [2m:[0;39m [Consumer
clientId=consumer-fooss-1, groupId=fo
浏览 1提问于2022-09-22得票数 1
回答已采纳
"idleBetweenPolls“是spring-kafka中的一个方便的配置，用于调整消费速率。 然而，当检查springboot的源代码和文档时，似乎没有提到这个道具，也没有提到如何将一些特定的值传播到org.springframework.kafka.listener.ContainerProperties.中 有没有人知道如何实现上述目标？
浏览 261提问于2020-06-19得票数 0
我使用的是Axon版本(3.3)，它在SpringBoot主类中使用
@SpringBootApplication(exclude = KafkaAutoConfiguration.class)
在我们的用例中，命令端微服务需要从kafka主题中挑选消息，而不是将其公开为Rest api。它会将事件存储在事件存储中，然后将其移动到另一个kafka主题，供查询端微服务消费。
由于KafkaAutoCOnfiguration被禁用，我不能使用spring-kafka配置来编写消费者。如何在Axon中使用普通消息？
我试着写了一个普通的Kafka spring Consumer，但由于Kafka A
浏览 18提问于2019-07-12得票数 0
我正在做一个POC来演示一些Spring Cloud Kafka功能。我使用的是带有SpringBoot 2.4.1的Java11。我的build.gradle具有以下库
implementation 'org.apache.kafka:kafka-streams'
implementation 'org.springframework.cloud:spring-cloud-stream'
implementation 'org.springframework.cloud:spring-cloud-stream-binder-ka
浏览 0提问于2021-01-30得票数 0
我对kafka技术很陌生，我需要从DB中获取所有实时数据，并将其传递给springboot微服务进行处理。在我的分析中发现，带有kafka源连接的apache可以将所有实时数据从数据库中提取到kafka主题。有人能告诉我们，有没有办法从kafka主题中选择这些数据并通过触发kafka服务的tell调用来共享到微服务？这个想法是，每当在数据库表中添加一个新条目时，kafka就可以通过kafka连接来提取数据，而且卡夫卡应该调用microservice并共享这个新条目。卡夫卡有可能吗？
数据库？？> kafka连接？？> kafka (主题)？？>一些称为微服务的服务？？>
浏览 3提问于2020-09-18得票数 0
我有一个可以批量消费消息的Kafka监听器。我也在使用手动提交。如果其中任何一条消息出现问题，我将使用返回语句，而不是提交偏移量(我希望再次处理所有这些消息)。 我的问题是- 在Kafka Listener中使用return语句有副作用吗？再次消费相同的消息需要多长时间？ 我正在使用Spring Kafka和SpringBoot。
浏览 34提问于2021-04-26得票数 0
回答已采纳
我必须发布来自kafka主题的消息到lambda来处理它们并使用springboot应用程序存储在数据库中，我做了一些研究并发现了一些可以使用kafka public Function<KStream<String, String>, KStream<String, String>> process(){}的消息，但我不确定这是否只用于将所消耗的消息发布到另一个kafka主题，还是可以用作一个事件源来进行lambda，我需要一些关于消费和将所消费的kafka消息转换为事件源的指导。
浏览 5提问于2022-08-11得票数 -2
我的目标是为Spring创建自己的骆驼组件。
我已经成功地创建了它并开始消费。我还希望能够停止组件和消费(使用JMX，使用其他Camel路由，.)，而不丢失任何消息。要做到这一点，当停止Camel组件时，我需要停止一个MessageListenerContainer，并最终停止在MessageListenerContainer中注册的MessageListener。
我的问题是，当MessageListenerContainer停止时，MessageListener仍然在处理消息。
@Override
protected void doStart() throws Except
浏览 9提问于2020-11-27得票数 0
回答已采纳
得到：
创建名为“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
回答已采纳
我开始使用Kubeflow，我读过关于KFServing的文章。这就是卡夫卡与卡夫卡从一个话题中消费的整合。我需要的是运行一个推理管道，它消耗来自Kafka的数据，进行一些数据预处理，运行ML预测，然后将结果发布回Kafka，如下所示：
我知道我可以用K本机事件来摄取卡夫卡主题1的数据。有什么方法可以用Kafka作为接收器吗?还是我需要编写一个独立的容器，将事件发布到Kafka主题2？
谢谢!
浏览 0提问于2020-03-03得票数 2
我对springboot kafka是个新手，我在这篇文章之后创建了一个例子。
我当前使用的是spring.kafka.version 1.1.6
我想向消息添加自定义标头，这样我就可以在标头中发送某些属性，例如: fileName、fileId
我发现你可以将kafka元数据设置为头部，但这并不能满足我的目的。
我有什么办法可以做到这一点吗？如果可能的话，如果你能分享一个例子，我将非常感谢。
浏览 12提问于2018-02-23得票数 0
我使用的是spring cloud stream kafka。
我正在尝试向kafka发送超过默认1000KB的数据。
我在server.properties文件中设置message.max.bytes，如下所示：
message.max.bytes=20000000
我的springboot应用程序中的application.properties如下所示：
spring.cloud.stream.kafka.bindings.<channelName>.producer.configuration.max.request.size=20000000
这是可行的。然而，我想概括一下
浏览 6提问于2019-08-28得票数 4
我使用SpringBoot1.5.6 Kerberos 来使用Kerberos身份验证连接到Kafka 0.11。这些是我为Kafka使用的依赖关系：
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-kafka</artifactId>
<version>3.0.0.RELEASE</ver
浏览 4提问于2019-02-20得票数 3
回答已采纳
我有一个springboot引导应用程序，我想通过JMX向Prometheus公开kafka.consumer指标。我可以看到度量标准，但我不知道在哪里设置这些mBeans(即kafka.customer:type= see mBeans，client-id=(-.\w+))。我了解到，只需要通过这个设置来激活JMX，但我不知道通过JMX向Prometheus公开这些指标需要做些什么。
浏览 9提问于2020-06-16得票数 1
回答已采纳
我一直在使用@KafkaListener作为一类Kafka事件消费者，它必须具有唯一的ids，并且每个事件使用者都必须有一个具有非常特殊名称的单独组。例如：
@KafkaListener(id="AmazingProductEventConsumer",
topics = "${kafka......topic}",
clientIdPrefix = "AmazingProductEventConsumerClientId",
groupId = "
浏览 10提问于2021-05-31得票数 0
这样的情况可能吗？
有一个应用程序"XYZ“(其中没有Kafka)公开了REST。它是一个SpringBoot应用程序，它与角应用程序进行通信。
创建了一个新的应用程序(SpringBoot)，它希望使用Kafka，并需要从"XYZ“应用程序中获取数据。它想用卡夫卡来做这个。
"XYZ“应用程序有一个示例端点[GET] api/message/all，它显示所有消息。
有没有一种方法可以将Kafka直接连接到这个端点并从中读取数据？简而言之，卡夫卡的想法是直接从EP消费数据。两个微服务之间的通信，其中一个微服务没有卡夫卡。
你对解决这种情况有什么建议。因为我想这个选择是
浏览 3提问于2022-02-05得票数 0
我在@StreamBuilder中使用KStream参数。
这将通过KStreamBinder创建一个KStreamBinder。
我的要求是使用绑定可视化和控制。
然而，
return new DefaultBinding <> (name, null, outboundBindTarget, null);
由于生命周期为null，所以不能通过springboot引导/执行器/绑定来控制状态。
(POST /actuator/bindings/{bindings-name} {"state":"PAUSED"})
如何控制绑定的状态？
我正在使用的
浏览 3提问于2020-02-18得票数 0
回答已采纳
我有一个Springboot应用程序。使用此配置：
kafka:
consumer:
listener:
ack-mode: MANUAL_IMMEDIATE
广告给这个消费者：
@KafkaListener(topics = "test", groupId = "group_id")
public void consume(String message, Acknowledgment ack) throws IOException {
ack.acknowledge();
}
但是
浏览 6提问于2020-03-30得票数 3
回答已采纳
我正在学习如何使用TestContainers测试Spring Boot Kafka应用程序。测试通过。然而，在一开始就有很多这样的消息： 2021-04-05 09:00:13.927 WARN 1864 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be
浏览 483提问于2021-04-05得票数 1
回答已采纳
我正在开发一个spring集成应用程序，它有一个Kafka出站通道适配器，并使用配置流。
弹簧-集成-核心:4.2.4
春季-整合-卡夫卡:1.3.0
spring-integration-java:1.1.2
我已经配置了消息处理程序规范，类似于下面的片段。
KafkaProducerMessageHandlerSpec messageHandlerSpec = Kafka
.outboundChannelAdapter()
.addProducer(new ProducerMetad
浏览 4提问于2016-02-01得票数 1
回答已采纳
我正在使用camel-spring-boot-starter运行一个kafka消费者应用程序。SpringBoot应用程序实现CommandLineRunner。以下属性是在我的应用程序中配置的。 camel.springboot.main-run-controller=true 我试图在端口8080上公开springboot执行器指标，但在将端点公开为非web应用程序时，我面临着挑战。url http://localhost:8080/actuator/metrics给出404状态代码。 我如何为我的用例公开HTTP端点上的致动器“/metrics”？ 注意:执行器和驼峰管理依赖项在类路径
浏览 15提问于2021-06-22得票数 0
我们的工作环境是一个被篡改的卡夫卡。我想知道在这种场景中部署kafka连接器和kafka流应用程序的最佳实践。目前，我们正在将每个连接器和流部署为springboot应用程序，并作为systemctl微服务启动。我不认为在对接每个卡夫卡连接器和流明显的优势。请给我提供同样的见解。
浏览 1提问于2018-02-20得票数 4
回答已采纳
我正在尝试在OpenShift中部署mongodb (1个主要的和2个次要的)，它们被部署了，但进入了“错误”状态，错误如下。 exception in initAndListen: IllegalOperation: Attempted to create a lock file on a read-only directory: /data/db, terminating 在发布以下adm策略后，mongo pod已成功启动并运行。 oc adm policy add-scc-to-group anyuid system:authenticated 现在我正在尝试部署一个连接到mongo
浏览 32提问于2020-01-09得票数 2
回答已采纳
1回答
我有一个springboot应用程序，它执行向表(Mysql)的插入操作。这反过来触发并插入到另一个表中。是否可以使用kafka从表中读取数据。
浏览 2提问于2018-10-31得票数 0
在已经部署在Stream中的SpringBoot-application中，我使用了yaml-file中的属性：
spring:
kafka:
producer:
retries: 8
是否可以在Spring Cloud DataFlow的UI中显示和管理该属性，如下所示：
浏览 0提问于2020-12-29得票数 0
我用了阿帕奇骆驼卡夫卡和弹簧靴
<camel.version>3.14.2</camel.version>
我在apache camel Kafka组件上使用了默认配置。
<dependency>
<groupId>org.apache.camel.springboot</groupId>
<artifactId>camel-kafka-starter</artifactId>
<version>${camel.version}</ver
浏览 3提问于2022-04-25得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
