Spring Kafka监听器线程名的含义？
Spring Kafka监听器线程名是指用于消费Kafka消息的线程的名称。在Spring Kafka中，可以使用ConcurrencyAwareMessageListener或MessageListenerContainer来创建监听器以接收和处理Kafka消息。
线程名的含义是为了标识和区分不同的线程。在Spring Kafka中，每个线程都会被分配一个独立的线程名。线程名通常由框架自动生成，并包含一些关键信息，如监听器容器的名称、分区信息等。通过线程名，可以方便地追踪和调试不同的线程操作。
监听器线程名的分类：根据不同的监听器容器和消费者配置，监听器线程名可以分为多个类别。例如，使用@KafkaListener注解时，可以通过concurrency属性配置并发消费者的数量，这将导致多个线程被创建，每个线程都有一个独立的线程名。
优势：线程名的存在可以帮助我们更好地理解和分析消费者的工作情况。通过观察线程名，我们可以知道每个线程正在消费的分区、消费者组的成员以及其他相关信息。这对于监控和故障排除非常有帮助。
应用场景：线程名通常在调试和故障排除期间使用。当我们需要跟踪和定位消息消费的问题时，可以通过观察线程名来确定哪些线程正在处理哪些消息。
推荐的腾讯云相关产品：腾讯云消息队列（Tencent Cloud Message Queue，CMQ）。腾讯云的CMQ是一种高可靠、可扩展、全托管的消息队列服务，适用于云原生应用开发、微服务架构、事件驱动架构等场景。CMQ提供了丰富的消息队列功能，可以与Spring Kafka等消息中间件很好地配合使用。具体产品介绍和使用指南可以参考腾讯云官方文档：腾讯云消息队列CMQ产品介绍。
相关·内容
我是Spring和Kafka的新手。我正在做一个使用SpringBoot-kafka的用例，允许用户在运行时创建kafka主题。spring应用程序期望在运行时以编程方式订阅这些主题。到目前为止，我所知道的是，Kafka listener是设计时，因此在启动之前需要指定主题。有没有办法在SpringBoot-Kafka集成中动态订阅kafka主题？
已引用此
我计划实现的当前方法是，不要使用Spring-Kafka集成，而是自己使用这里提到的代码实现Kafka消费者
浏览 0提问于2017-01-08得票数 11
当我在本地的机器上把卡夫卡旋转起来的时候，我就能让卡夫卡工作得很好。但是，当我试图让它在AWS实例上工作时，似乎没有什么是正确的。我试着旋转我自己的服务器，就像我在本地旋转动物园管理员和卡夫卡一样
curl http://apache.spinellicreations.com/kafka/0.10.0.0/kafka_2.11-0.10.0.0.tgz | tar -xzf
cd kafka_2.11-0.10.0.0
bin/zookeeper-server-start.sh config/zookeeper.properties &
bin/kafka-server-st
浏览 2提问于2016-10-20得票数 2
回答已采纳
1回答
我只想理解kafka rebalance.This是我的监听器方法。我已经将RetryTemplate与消费者工厂配置为重试20次，回退延迟为20秒。我使用的是spring-kafka 1.2.2(我们计划升级客户端)和手动ack。
@KafkaListener(id = "${kafka.listener-id}", topics = "${kafka.topic}")
public void listen(final ConsumerRecord<String, String> consumerRecord,
浏览 0提问于2018-09-24得票数 0
我使用的是kafka和spring cloud streams。我的侦听器代码如下所示 @Bean
public Consumer<List<Map<String, Object>>> receive() throws MyException {
return message -> {
try {
messageService.processMessage(message);
} catch (MyException e) {
浏览 158提问于2020-04-03得票数 2
1回答
我已经为我的基于spring boot的库配置了以下Kafka属性，这些库捆绑在部署到Wildfly的EAR的lib目录中。通过从类路径加载porperty文件(WEB-INF/classes)，我能够成功地启动spring组件。 spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.gson.GsonAutoConfiguration,org.springframework.boot.autoconfigure.jms.artemis.ArtemisAutoConfiguration,\
org.spring
浏览 102提问于2021-04-28得票数 1
回答已采纳
1回答
在卡夫卡消费者方面，我一直面临以下例外。令人惊讶的是，这个问题是不一致的和代码的旧版本(配置完全相同，但一些新的无关特性)工作正常。有人能帮我确定是什么原因造成的吗？
[ERROR][938f3c68-f481-4224-b2c6-43af5fb27ada-0-C-1][org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer] - Error handler threw an exception
org.springframework.kafka.KafkaException: Seek
浏览 8提问于2021-06-11得票数 0
回答已采纳
1回答
我使用的是bitnami/kafka:最新的和bitnami/zookeper:最新的图片。
我在Github中的示例项目：
如果您运行命令docker-compose，然后运行应用程序，您将看到Spring应用程序无法连接到kafka容器(该容器目前运行在主机kafka:9092下)。
来自应用程序的堆栈跟踪：
2022-05-31 08:59:18.255 WARN 144414 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-foo-1, group
浏览 13提问于2022-05-31得票数 0
回答已采纳
我想做什么
我有一个用Flask编写的HTTP服务，它是一个用于构建不同服务实例的模板。因此，这个模板需要是可泛化的，以处理实际使用和不包含Kafka消费的用例。
我的目标是让一个可选的Kafka消费者在API模板的背景下运行。我希望任何需要它的服务都能够异步地从Kafka主题读取数据，同时也像通常那样独立地响应HTTP请求。这两个进程(Kafka消费，HTTP请求处理)并不相关，只是它们将发生在同一服务的幕后。
我写的东西
这是我的装置：
# ./create_app.py
from flask_socketio import SocketIO
socketio = None
def
浏览 4提问于2019-10-23得票数 0
1回答
春天-卡夫卡听众同意
、、、
我使用实现了Kafka消费者。我有一个带有2个分区的Kafka主题，并且我使用了并发级别设置为2的ConcurrentKafkaListenerContainerFactory，因此每个容器实例都应该按照spring 从单个分区中使用。
KafkaMessageListenerContainer在一个线程上接收来自所有主题/分区的所有消息。ConcurrentMessageListenerContainer委托一个或多个KafkaMessageListenerContainer来提供多线程消耗。
这是我的消费类：
@Component
public class KafkaConsume
浏览 3提问于2017-09-21得票数 5
回答已采纳
使用Spring JDBC模板，您可以使用一个简单的标志来懒惰地初始化连接。对于Springboot 1.5.x / Spring Kafka 1.3.x部署的Kafka容器工厂，是否有类似的功能？
到目前为止，我看到的最好的答案是禁用autoStart并自己管理启动，在这里捕获启动过程中可能发生的任何异常-
这是唯一的方法吗?在使用KafkaListenerEndpointRegistry自我管理容器的生命周期时，是否有任何注意事项？
Lazy注解是否适用于@KafkaListener，@Configuration类的Kafka配置或类似的组件类？提出这个问题，因为似乎没有一个书面的方法来
浏览 0提问于2019-10-29得票数 1
我们使用Spring Kafka 2.2.2版本通过@KafkaListener和ConcurrentKafkaListenerContainerFactory从Kafka中检索记录。我们已经将max-poll-records配置为5，但是它始终只向消费者提供列表中的1条记录，而不是5条记录。
虽然使用相同的配置，但它可以在Spring Kafka 2.1.4版本中工作。
下面是我们的application.yml配置：
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
enable-auto-
浏览 0提问于2019-06-23得票数 1
2回答
一个简单的spring kafka，它使用的是网络集群上的一个主题：
错误：
无法建立与节点-1 (localhost/127.0.0.1:9092)断开连接的引导代理本地主机:9092 (id：-1齿条: null)。可能找不到经纪人。
谜题：
配置的代理是而不是本地的，它是BROKER_1.FOO.NET:9094，并且是可用的。
pom.xml
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</
浏览 7提问于2021-06-08得票数 1
回答已采纳
我查看了这些文档，发现如下：
spring.kafka.producer.properties.spring.json.type.mapping=cat:com.mycat.Cat,hat:com.myhat.Hat
我的制片人的财产：
spring.kafka.producer.bootstrap-server=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-seriali
浏览 4提问于2022-09-16得票数 0
回答已采纳
弹簧卡夫卡配置bootstrap.yml不工作。
我用的是application.yml而不是bootstrap.yml。现在起作用了。
spring:
kafka:
bootstrap-servers: localhost:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
浏览 9提问于2019-11-04得票数 1
1回答
刚开始使用Spring (2.1.4.RELEASE)和Kafka (1.0.0)，但是当我添加事务时，处理速度大大降低。
代码：
spring.kafka.consumer.max-poll-records=10
spring.kafka.consumer.specific.avro.reader=true
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.group-id=${application.name}
spring.kafka.consumer.properties.isolation.l
浏览 0提问于2018-04-10得票数 2
回答已采纳
我有spring kafka消费者应用程序。应用程序充当传递，轮询来自kafka代理的消息并将其发送到IBM MQ。在发生故障时存储偏移量的最佳/简单化方法是什么？
浏览 1提问于2019-03-24得票数 0
我试图在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
回答已采纳
1回答
如何在Spring Boot (2.2+)和Spring Kafka (2.4+)应用程序的@SpringBootTest测试中禁用@KafkaListener实例？ 目标是在特定测试中禁用Kafka侦听器，以便在不启动嵌入式Kafka代理的情况下运行此类测试。
浏览 39提问于2020-01-25得票数 3
回答已采纳
我有下面的docker compose yml， version: '3.5'
services:
zookeeper:
container_name: kafka-zookeeper
image: wurstmeister/zookeeper:3.4.6
ports:
- "2181:2181"
networks:
- my-network
kafka:
container_name: kafka-server
image: wurstmeister/kafka:2.12-2
浏览 21提问于2020-09-06得票数 0
回答已采纳
1回答
如何停止消息监听器内部的容器？
、、、、
我的回答是这样的 Spring for Kafka 2.3 setting an offset during runtime for specific listener with KafkaMessageListenerContainer 如果我收到一定的偏移量，我想停止容器。如何停止"onMessage“内部的容器？ 另外，是否可以为容器设置初始偏移量？
浏览 20提问于2021-04-28得票数 0
我正在开发一个微服务，其中我需要使用spring kafka ReplyingKafkaTemplate查询两个不同的微服务。
当我尝试使用一个微服务时，它可以很好地工作，只提供一个回复主题，如下所示
@Bean
public ReplyingKafkaTemplate<String, byte[], byte[]> replyKafkaTemplate(ProducerFactory<String, byte[]> pf,
KafkaMessageListenerContainer<String, byte[]> container) {
浏览 0提问于2018-10-22得票数 0
2回答
我有两个服务(都是kotlin上的spring boot )。命名为“客户端”和“服务器”由于一些限制，我必须使用同步请求-回复模式与kafka。所以我尝试使用ReplyingKafkaTemplate。我的问题是我需要使用多个实体。意味着创建多个ReplyingKafkaTemplate，一个用于"FOO“，第二个用于"BAR”。因此，在我的代码中，我创建了多个KafkaConfig类，并设置了每个实体和一个基本配置。
此外，我还从加载中排除KafkaAutoConfig。下面是"server“端的配置(不带kreplying kafka模板)：
@Configura
浏览 16提问于2020-05-26得票数 0
我们的基础设施必须管理两个spring boot应用程序之间的信息交换。某些信息基于Oracle DB。我们使用Kafka来通知哪些信息必须由接收方管理。
(注意:我必须泛化代码)
我们有Kafka producer，配置如下：
@Configuration
@Lazy(false)
public class KafkaConfig {
//...
@Bean(name="trManagerJpaKafka")
public ChainedKafkaTransactionManager<Object, Object> chainedTm(
浏览 7提问于2021-10-28得票数 0
我试图从同一个Kafka主题反序列化不同的JSON有效负载。这里提出的其他问题，引导我进行了第一次尝试，但我无法使它运行。
正如加里提到的()，有一些提示(JsonSerializer.ADD_TYPE_INFO_HEADERS)，但是当我发送和接收这两条消息时，我会得到一个异常。
org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
浏览 0提问于2019-01-24得票数 5
回答已采纳
所以我的问题是，我在我的属性文件中定义了一些Kafka主题，并且我可以在我的SpringBoot应用程序中从该主题读取KafkaStream<String, String>。但是我想访问KafkaStreams对象，这样我就可以打印我的KafkaStreams拓扑，这对开发很有用。在我的一个@StreamListener中，我尝试检索流生成器进程bean，这样我就可以以这种方式获取底层的KafkaStreams对象(如这里所描述的：https://cloud.spring.io/spring-cloud-stream-binder-kafka/spring-cloud-strea
浏览 46提问于2020-07-08得票数 0
1回答
我使用的是Spring Kafka Consumer。我已经将并发设置为10，并创建了5个消费者(针对5个主题)。所以有50个Spring Kafka消费者线程。 我可以为Kafka消费者提供的最大线程数是多少？如何增加这个线程池的大小？我已经看过了spring文档，但没有发现任何相关内容。
浏览 84提问于2021-08-17得票数 1
回答已采纳
我是spring cloud stream和kafka的新手。
如果外部应用程序没有得到最后一条消息，那么阻止消费者提交下一个偏移量的最佳方法是什么？我有一个中间件来管理使用者和外部应用程序之间的通信。如果出了什么问题，我想至少尝试发送消息10次，其间(到外部应用程序)有1分钟的延迟。
Middleware是一个简单的post请求应用程序，用于过滤kafka消息中的特定数据。
浏览 3提问于2019-10-24得票数 0
我查看了这些文档，发现如下：
spring.kafka.producer.properties.spring.json.type.mapping=cat:com.mycat.Cat,hat:com.myhat.Hat
我的制片人的财产：
spring.kafka.producer.bootstrap-server=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-seriali
浏览 27提问于2022-09-16得票数 0
回答已采纳
这是我的用例。 我有一个每天结束的提要进程，它在一个单独的进程上按计划运行。 在白天，我将通过Kafka主题接收实时更新，Spring Kafka监听器将消费和处理这些内容。但是，我想在EOD提要作业启动期间暂停Kafka侦听器，并在作业完成后恢复。 以下是我正在考虑的方法。 出于某些管理目的，我已经公开了JMX bean来停止和启动侦听器。当EOD提要工作开始时，我也想利用它来停止/启动Kafka侦听器。这将得到更同步的控制。 因为我对EOD工作有了更多的控制，什么时候开始，什么时候完成。在此期间，Kafka监听器可以根据时间表暂停。由于这是异步的，由于某些原因，这将导致问题EOD提要作业
浏览 76提问于2021-03-02得票数 0
回答已采纳
我使用绑定的基础设施进行流处理。
以下是主要的KafkaStream方法：
private static final String READ_STORE_NAME = "readMessageStore";
@Bean
public Consumer<KStream<String, ContentModel>> filter() {
return in -> in.groupByKey(Grouped.with(Serdes.String(), new JsonSerde<>(ContentModel.class)))
浏览 1提问于2021-05-12得票数 2
我想停止一个应用程序有一个KafkaListener后，一段时间没有收到消息(例如:2000ms)或在一个特定的时间已经过去了，因为它是第一次启动。
该应用程序是一个加载spring和kafka库的JAva应用程序。不使用SPRING BOOT
我还没有找到完全像这样工作的例子。你能帮忙吗？
浏览 3提问于2018-08-17得票数 0
2回答
我正在开发一个应用程序，它将在使用tomcat的spring引导上下文下运行。这个项目在启动时连接到Kafka，提取消息并处理它们。tomcat用于一些管理API。
当Spring加载时，它通过几个线程与Kafka连接。每个线程都带有一个处理消息的spring。这颗豆子当然会长出更多的豆子等等..
我需要考虑并发问题吗？同步的东西？例如，使用并发类型，例如ConcurrentHashMap，而不是普通的HashMap？
浏览 2提问于2016-01-12得票数 7
回答已采纳
1回答
我已经在集群中配置了3个kafka，我正在尝试使用spring。
但在我杀了卡夫卡领导人之后，我无法发送其他信息来排队。
我正在将Spring.kafka.boot-server属性设置为："kafka-1:9092;kafka-2:9093，kafka-3:9094“以及我的主机文件中的所有名称。
Kafka版0.10
有些人知道如何正确配置？
编辑
我已经测试了一件事，并发生了一种奇怪的行为。启动服务时，我会向主题发送一条消息(以强制创建)
代码：
@Bean
public KafkaSyncListener synchronousListener(MessageSender se
浏览 2提问于2018-06-11得票数 6
2回答
我在多线程feature(ConcurrentKafkaListenerContainerFactory)，中使用了spring，我发现了两种类型的线程名: 1. #0-1- kafka -this 12.0-1-kafka-listener-3。
那么，我如何理解这两种线程呢？他们之间有什么关系？提前感谢！
浏览 4提问于2017-05-18得票数 3
回答已采纳
我们使用Kafka 0.10.2运行Kubernetes集群。在集群中，我们有一个由10个副本组成的副本集，这些副本运行我们的一个服务，这些副本将一个主题作为一个消费者组进行消费。
最近我们打开了这个副本集的自动缩放功能，因此它可以根据CPU消耗增加或减少pod的数量。
在启用此功能后不久，我们开始在我们的Kafka队列中看到滞后。我查看了日志，看到消费者经常(几乎每隔5分钟)将协调器标记为已死，并在几秒钟后重新连接到同一协调器。
我还经常在日志中看到：
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot
浏览 1提问于2018-11-28得票数 1
我有一个@StreamListener方法，它将执行REST调用。当REST调用返回异常时，@StreamListener方法抛出RunTimeException并重试。@StreamListener方法抛出RuntimeException时，重试次数不限
Spring云流重试配置：
spring.cloud.stream.kafka.bindings.inputChannel.consumer.enableDlq=true
spring.cloud.stream.bindings.inputChannel.consumer.maxAttempts=3
spring.cloud.stream.
浏览 0提问于2018-08-27得票数 1
我正在尝试在Docker和Spring上配置Kafka。
spring应用程序负责生成消息。Docker应该使用它们。我已经准备好了容器，我正在尝试连接。
不幸的是，我得到一个错误：
[Producer clientId=1-1] 1 partitions have leader brokers without a matching listener, including [myTopic-0]
我的docker-component
version: '3'
services:
zookeeper:
container_name: zookeeper
ima
浏览 1提问于2020-06-26得票数 1
我们使用Spring Cloud Stream Kafka Binder (带项目反应堆集成，即Flux streams)和manual offset commits (即autoCommitOffset = false)。 我们正在尝试编写一个与Embedded Kafka from spring-kafka-test的集成测试，通过手动读取消费者组偏移量，使用admin client，在测试向我们的主题发送消息之前和之后，断言所有这些都是有效的。 测试间歇性地失败。使用awaitility，我们现在等待10秒来轮询偏移量，这似乎绕过了我们的大多数问题，因为偏移量将在大约7秒后发生变化-但
浏览 15提问于2019-07-01得票数 1
回答已采纳
1回答
我用Spring模块创建了一个卡夫卡听众。当我使用JMX打倒所有Kafka侦听器时，Springboot应用程序就会崩溃。在这个应用程序中，除了Kafka使用者之外，我没有其他线程在运行。
浏览 11提问于2021-02-23得票数 0
回答已采纳
1回答
我尝试用spring引导JAVA自动在apache上创建一个主题。为此，我遵循了本教程：，但我设法连接到服务器，但我不能创建一个主题。KAFKA服务器托管在一个EC2 AWS实例上。
以下是Intellij控制台中获得的错误的一部分
022-05-08 10:21:22.889 WARN 20424 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient : [AdminClient clientId=adminclient-1] Error connecting to node ip-172-31-XX-XX.eu-we
浏览 6提问于2022-05-08得票数 0
2回答
在我的Spring /Kafka项目中，我有以下的使用者配置：
@Configuration
public class KafkaConsumerConfig {
@Bean
public ConsumerFactory<String, String> consumerFactory(KafkaProperties kafkaProperties) {
return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties(), new Strin
浏览 0提问于2018-06-17得票数 2
回答已采纳
我想用创建一个Kafka流应用程序，它集成了两个不同的Kafka集群/设置。我尝试使用多绑定器配置来实现它，如文档中提到的那样，并且类似于下面的示例：。
给定这样一个简单的函数：
@Bean
public Function<KStream<String, AnalyticsEvent>, KStream<String, UpdateEvent>> analyticsEventProcessor() {
return input -> input
.filter(new AnalyticsPredicate())
浏览 2提问于2022-08-10得票数 0
回答已采纳
1回答
我使用和Kafka绑定。我可以看到，一旦应用程序启动，它就会在每分钟为我的应用程序中配置的所有输入绑定设置之后抛出信息级别日志。
application.properties中的配置
spring.cloud.function.definition=consumeMessage
spring.cloud.stream.bindings.consumeMessage-in-0.destination=Kafka-stream
spring.cloud.stream.bindings.consumeMessage-in-0.group=Kafka-stream-consumer-group
原木是
浏览 1提问于2021-06-25得票数 2
回答已采纳
1回答
我目前正在尝试编写一个适配器，它将使用来自ActiveMQ的消息并将其发布到Kafka。
我正在考虑使用spring集成来集成这两个消息传递系统。
我的问题是，我的应用程序不会维护模型的注册表，许多应用程序将使用这些模型将记录发布到activeMQ。我想要接收这些javax消息，并想执行一些转换，比如将jmscorrelationId添加到kafka消息中。
ALso，另一个要求是仅当kafka发送/发布成功时才向活动mq发送确认。
是否可以使用spring集成将ack发送回activemq？
spring集成是个不错的选择吗？
请注意，我的技术架构师不赞成使用Camel/Mule。此外，他不
浏览 10提问于2017-07-05得票数 0
回答已采纳
我在docker网络中运行Apache Kafka。Kafka代理的容器称为docker_kafka。我想要的是允许从Kafka客户端从另一个码头容器以及从码头主机连接。docker_kafka的端口9092已映射到坞站主机。
我一直在尝试KAFKA_LISTENERS和KAFKA_ADVERTISED_LISTENERS环境变量。例如，当我使用KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092时，另一个docker kafka客户端无法通过PLAINTEXT://docker_kafka:9092连接到kafka，反之亦然。它不允许多个
浏览 1提问于2017-03-29得票数 4
嗨，我们一直在使用旧的spring版本和kafka 1.1，有以下的依赖项
+--- org.springframework.boot:spring-boot-starter-web: -> 1.5.4.RELEASE (*)
+--- org.springframework.cloud:spring-cloud-starter-config: -> 1.3.1.RELEASE (*)
+--- org.springframework.cloud:spring-cloud-starter-bus-kafka: -> 1.3.1.RELEASE (*)
+--- org.spr
浏览 2提问于2020-05-20得票数 0
我现在正在玩Spring Boot v2。到目前为止，我的设置是这样的: Spring -> Telegraf -> Kafka -> Telegraf -> influx
我想知道是否有可能在春天和卡夫卡之间拿出第一个telegraf，所以就像这样: Spring -> Kafka -> Telegraf -> Influx
我已经查看了测微器的配置，但没有为Kafka配置。此外，telegraf正在从Spring中提取数据。由于Kafka是一个推送模型(即，您正在将数据推送到Kafka中)，Spring是否能够将数据推送到Kafka？如果是，是如何
浏览 0提问于2019-02-12得票数 0
我试图使用手动确认从一个主题消费多个消息，但如果所有消息都只有一次确认，则确认工作。
@KafkaListener(
id = "${kafka.buyers.product-sales-pricing.id}",
topics = "${kafka.buyers.product-sales-pricing.topic}",
groupId = "${kafka.buyers.group-id}",
concurrency = "${kafka.buyers.concurrency}"
)
publ
浏览 4提问于2019-08-01得票数 1
据我所知，如果我在同一进程中有topic1=ClassA，topic2=ClassB，我需要两个容器工厂？
我的配置类：
@Bean
public ConsumerFactory<String, MessageADto> xxxConsumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), new JsonDeserializer<>(MessageADto.class));
}
@Bean
publi
浏览 0提问于2020-02-03得票数 0
我正在从数据库中提取数据，以检查是否有可用的系统资源来处理来自KafkaListener的进一步消息。如果我的条件不满足，我想让@KafkaListener暂停，当条件满足时，我想让@KafkaListener继续。如何在spring Kafka中实现这一点？
另外，暂停某个特定分区的使用者有什么缺点吗？
浏览 0提问于2017-07-25得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
云直播
对象存储活动推荐
腾讯云开发者
