Kafka消费者提交线程安全
是指在使用Kafka消息队列时，消费者提交偏移量的操作是线程安全的。具体来说，当多个消费者同时提交偏移量时，Kafka会确保这些提交操作不会相互干扰，保证数据的一致性和可靠性。
Kafka是一个分布式流处理平台，具有高吞吐量、可扩展性和容错性等特点。它通过将消息分区存储在多个服务器上，实现了消息的持久化和高效的消息传输。在Kafka中，消费者可以通过提交偏移量来记录已经消费的消息位置，以便在发生故障或重启后能够继续从上次的位置消费消息。
Kafka消费者提交线程安全的优势在于：
- 数据一致性：多个消费者可以同时提交偏移量，而不会出现数据冲突或丢失的情况。Kafka会保证提交操作的原子性和顺序性，确保消费者提交的偏移量是准确的。
- 高可靠性：即使在消费者发生故障或重启的情况下，Kafka也能够保证消费者能够从上次提交的偏移量位置继续消费消息，避免数据重复消费或丢失。
- 并发处理：由于Kafka消费者提交线程安全，多个消费者可以并发地提交偏移量，提高了消息处理的并发性能和效率。
Kafka消费者提交线程安全的应用场景包括但不限于：
- 实时数据处理：在实时数据处理场景下，多个消费者可以同时提交偏移量，确保数据的实时性和准确性。
- 日志收集与分析：Kafka常用于日志收集与分析系统中，多个消费者可以并发地提交偏移量，实现对大量日志数据的高效处理和分析。
- 消息队列应用：Kafka作为消息队列的一种实现方式，多个消费者可以同时提交偏移量，实现消息的可靠传输和处理。
腾讯云提供了一系列与Kafka相关的产品和服务，包括云原生消息队列 CMQ、消息队列 CKafka、云消息队列 CMQ for Kafka 等。这些产品可以满足不同场景下的消息传输和处理需求。您可以通过以下链接了解更多关于腾讯云的相关产品和服务：
相关·内容
我正在尝试从最早的偏移量中读取卡夫卡主题，然后通过python脚本对某些记录进行墓碑。由于消息数量巨大(百万+)，我希望利用多处理来使脚本在使用消息时更快。下面是脚本中的一段代码：
from kafka import KafkaConsumer
def cleanup_kafka_topic(self, env):
# Declarations
consumer = KafkaConsumer(<topic_name>, group_id=<some_group>),
boots
浏览 71提问于2021-02-19得票数 0
kafka使用者无法仅提交特定分区的偏移量。
aklsfoipafasldmaknfa asiofuasofiusaofasd
[2019-01-04 12:22:22,691] ERROR [Consumer clientId=consumer-1, groupId=console-consumer-11955] Offset commit failed on partition my-topic-2-9 at offset 0: The request timed out. (org.apache.kafka.clients.consumer.internals.ConsumerCo
浏览 3提问于2019-01-04得票数 5
我有一个有多个主题的Kafka集群，我将为每个主题设置一个分区，所有这些主题将由一个运行3个Kafka消费者线程的EC2实例消耗(每个线程一个消费者)，属于同一个消费者组。
我还没有尝试过，但我想知道Kafka是否可以平衡所有主题的分区，使其平均由3个线程消耗？或者Kafka会将所有分区都分配给一个线程使用？
浏览 1提问于2019-01-16得票数 1
Kafka是否允许一个线程或进程使用来自分区的数据，而另一个线程或进程则负责在数据完全处理后手动提交偏移量？
浏览 0提问于2018-04-17得票数 1
假设我们有一个Kafka Sink连接器从主题TopicA中填充一个数据库，
卡夫卡流应用程序推动了对TopicA主题的更新。
我想知道是否：
由Sink连接器更新到数据库中的数据，以及执行更新的
受Kafka流应用程序提交间隔的影响，如果是，如何影响？
浏览 4提问于2020-04-25得票数 0
回答已采纳
1回答
使用Spring Boot Kafka -v 2.2.1.RELEASE (使用kafka-clients.jar v 2.3.1) 我正在为我的Kafka代理使用confluent v5.1.2。 我已经启用了自动提交为true，并且每隔3秒自动提交一次。 我有一个基于spring的监听器类(@KafkaListener)，每个消息处理大约需要6-7秒，因为它是一个资源密集型操作。 我在这里注意到的是，监听器从未提交过偏移量，并且延迟永久地保持为"550“。 正因为如此(我认为)，消息被监听器无限期地处理。 当我将侦听器更改为手动提交偏移量时，它工作得很好。 为什么在涉及长时间运行
浏览 20提问于2020-01-23得票数 0
我使用手动kafka commit，将属性enable.auto.commit设置为false，同时初始化Kafka消费者，并在接收和处理消息后手动调用kafka commit。
然而，因为在我的消费者中处理消息是很耗时的，所以我得到了Exception with message "error": "Broker: Group rebalance in progress"
原因是重新平衡超时后提交被拒绝，并出现此错误。现在，恢复操作是退出并重新实例化进程，这将再次触发重新平衡和分区分配。另一种方法是捕获此异常，然后照常继续，只有在poll()调用被阻塞直到重新平
浏览 15提问于2020-05-19得票数 0
我正在运行下面的代码，在从Kafka读取了几个RDDs之后，我得到了这个Kafka多线程访问错误。不确定是什么导致了这种情况。基本上，代码所做的是从kafka读取CDR，并尝试将其放入全局临时视图中。简而言之，所有的RDDs /数据帧都必须存储在全局视图中。我使用的是Kafka 0.10和spark 2.1.1
val schema_string = "subscriberNumber, originNodeType, originHostName, originOperatorID, originTimeStamp, currentServiceClass, voucherBase
浏览 0提问于2018-05-06得票数 1
1回答
我们有一个flask应用程序，在gevent worker模式下使用gunicorn进行部署。该应用程序本身侦听很少的REST app，但也有一个kafka消费者(使用confluent客户端)侦听。当我们部署应用程序时，kafka消费者会按预期工作，但当向应用程序提交REST请求时，应用程序不会处理这些请求。 看起来应用程序线程卡住了kafka消费者。如果我们删除消费者代码，应用程序将按预期工作。 有什么建议吗？
浏览 150提问于2020-01-02得票数 1
回答已采纳
1回答
我理解kafka将主题的数据安排在多个分区上，并且一个消费者组中的消费者被分配了不同的分区，他们可以在那里接收数据：
partition-1 partition-2 partition-3 partition-4 partition-5 partition-6
|____________| |____________| |____________|
| | |
consumer-1
浏览 6提问于2021-10-22得票数 0
我们正在使用低级处理器API开发一个Kafka Streams应用程序。 根据Kafka上的文档，所有的线程和并行性都是由Stream线程和流任务处理的。使用主题上的分区，并行性也是可扩展的。 当前代码如下所示： public class Processor implements Processor<K, V> {
@Override
public void process(String key, V value) {
//Do processing on the stream thread itself
...
// Write b
浏览 15提问于2019-05-29得票数 2
我使用的消费者组只有一个消费者，只有一个经纪人( docker wurstmeister image )。在代码中决定是否提交偏移量-如果代码返回错误，则不提交消息。我需要确保系统不会丢失任何消息-即使这意味着永远重试相同的消息(现在;)。为了测试这一点，我创建了简单的处理程序，它不会在'error‘字符串作为消息发送到kafka的情况下提交偏移量。所有其他字符串都被提交。
kafka-console-producer --broker-list localhost:9092 --topic test
>this will be commited
现在正在运行
kafka-run
浏览 1提问于2018-03-19得票数 2
当使用时，我可以通过设置我的kafka生产者的kafka.compression.codec属性来设置编解码器。
假设我在我的生产者中使用了snappy压缩，当我使用某个kafka消费者消费kafka的消息时，我应该做一些事情来解码来自snappy的数据还是它是kafka消费者的内置功能？
在中，我找不到任何与kafka consumer中的编码相关的属性(它只与生产者相关)。
有人能把这个清理一下吗？
浏览 0提问于2013-11-10得票数 9
回答已采纳
我正在使用kafka 0.9.0.1 broker和0.9.0.1客户端。我的使用者实例正在消耗处理时间小于1秒的记录。和其他主要的秘密是
enable.auto.commit=false
session.timeout.ms=30000
heartbeat.interval.ms=25000
处理后，我将提交偏移。我得到了例外
为组UNKNOWN_MEMBER_ID提交偏移时发生错误kafka_to_s3
错误com.bsb.hike.analytics.consumer.Consumer -由于组重新平衡，无法提交retryCount=2 retryCount=2提交无法完成
一小
浏览 1提问于2016-09-28得票数 2
1回答
我试图找到在Kafka消费者中手动提交偏移的方法，使用Spring (1.1.0.RELEASE)。我理解，更好的做法是为健壮的客户端实现提交这些抵消，这样其他使用者就不会处理重复的事件，这些事件原本可能是由已经死了的使用者处理的，或者是因为触发了再平衡。
我知道有两种方法-
将ACK_MODE设置为MANUAL_IMMEDIATE，并在侦听器实现中调用ack.acknowledge() API
ConcurrentKafkaListenerContainerFactory.getContainerProperties().setAckMode(AbstractMessageListene
浏览 1提问于2018-06-13得票数 3
我提交了多个spark streaming作业，这些作业使用相同的group.id消耗同一个kafka topic，正如纯Kafka文档中所说的，拥有相同group.id的多个消费者将加入同一个消费组，并且kafka topic中分块的记录将被拆分成这些消费者。然而，正如我在我的工作中测试的那样，这两个spark streaming作业仍然使用相同的主题的所有分区(不是均匀分割的)，并且在who进程期间没有发生重新分区/异常。这里有人知道Spark是如何管理与纯kafka平台不同的kafka分区偏移的吗？也许这是因为zookeeper管理spark中的kafka偏移量，而不是kafka本身管
浏览 63提问于2021-07-30得票数 0
我对Kafka很陌生，我有一个使用java微服务的kafka流，它消耗了由生产者和过程产生的kafka主题的信息。使用auto.commit.interval.ms设置了kafka提交间隔。我的问题是，在提交之前，如果微服务崩溃，那么处理但未提交的消息会发生什么情况？会有复制的记录吗？如果发生这种重复，该如何解决？
浏览 6提问于2022-02-25得票数 1
3回答
Kafka消费者崩溃
、、、
我的应用程序使用来自Kafka 0.9的数据。
目前，我们获取消息、处理和提交。在此过程中，如果使用者由于处理时间过长而无法发送heartBeat，使用者协调器将假定使用者已死，并重新平衡分区。因此，总体来说，这导致了消费者数量的减少。过了一段时间，我们的数据处理就停止了。
我应该如何处理这种应用程序故障？
如果协调器发现一个新的消费者已经死了，有没有办法让它活着或者跨越一个新的消费者呢？
浏览 0提问于2020-02-27得票数 1
我们使用的是之前版本的Kafka(0.8.X)，其中所有偏移量细节都存储在ZooKeeper中。现在我们转向了卡夫卡(0.10.X)的新版本，所有的主题偏移细节都存储在Kafka本身中。我们必须将所有的主题偏移细节转移到ZooKeeper，以获取生产中现有的应用程序。卡夫卡安装在Windows机器上。我们不能从窗口运行kafka-消费群。请建议如何将提交的偏移量从ZooKeeper迁移到Kafka。
浏览 0提问于2018-02-08得票数 0
我使用了下面的Kafka消费者示例代码：
KafkaConsumer kconsumer = new KafkaConsumer(props);
ConsumerRecords<String, byte[]> records = kconsumer.poll(timeoutInMS);
kconsumer.commitSync();
我有两个问题：
(1)偏移量提交的地方是Kafka还是Zookeer？(我知道旧的kafka版本是提交给zookeeper的，但是新的kafka版本仍然是ZK，我甚至没有在配置文件中指明zookeeper的IP )。
(2)如果需要重复使用数据，如何检
浏览 4提问于2018-01-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
在我的应用程序中，有多个企业。每个企业登录并执行一些操作，如上传数据，然后Kafka producer获取数据并发送到主题。另一方面，Kafka消费者消费来自主题的数据并执行业务逻辑。并持久保存到数据库中。在这种情况下，当单个企业登录时，一切都是完美的。但当多个企业登录时，Kafka将按顺序消费。也就是说，我怎样才能使进程并行？在多个客户端请求上。提前谢谢。
浏览 2提问于2019-02-12得票数 0
我使用的是spring kafka，我想知道kafka消费者什么时候会被驱逐出群。当处理时间超过轮询间隔时，它会被逐出吗？如果是，那么心跳的目的不是表明消费者是活的，如果发生这种情况，那么消费者永远不应该被驱逐，除非进程本身失败。
浏览 1提问于2020-05-22得票数 0
5回答
我第一次使用Spring，我无法在我的消费者代码中使用Acknowledgement.acknowledge()方法进行手动提交。如果我的使用者配置或侦听器代码中缺少什么，请让我知道。否则，是否有其他方法处理基于条件的确认偏移。在这里，我看到的解决方案是，如果偏移量没有手动提交/确认，它应该由使用者选择相同的消息/偏移。
配置
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.commo
浏览 2提问于2017-11-22得票数 37
我处理活动应用程序的时序数据。因此，旧数据没有任何意义。我只想处理流应用程序启动后收到的数据，而不是之前提交的偏移量。重启后忽略kafka流app旧记录的正确方式是什么？
对于kafka consumer API，我通常使用seekToEnd()方法来跳转到最新的记录。是否有用于流的等效机制？我希望避免过滤自上次提交以来的所有消息，以忽略旧消息。
浏览 3提问于2017-11-15得票数 2
卡夫卡事件被反复回放给消费者。我可以看到下面的例外-
5-Nov-2019 10:43:25 ERROR [org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run : 685] :: org.springframework.kafka.KafkaListenerEndpointContainer#2-0-C-1 :: :: Container exception
org.apache.kafka.clients.consumer.CommitFailedException:
浏览 10提问于2019-11-26得票数 0
我的Kafka Producer正在以大约..的速度生成消息。每个30 seconds的350 mb。
Kafka Setup
--> 1 Zookeeper instance
--> 3 Kafka Brokers
--> 1 Java Producer
--> 1 Java Consumer
这就是我创建Topic and broker partitions的方式
bin/kafka-topics.sh --create --zookeeper 10.10.1.5:2181 --replication-factor 1 --partitions 8 --topic
浏览 1提问于2014-04-19得票数 1
1回答
我有一个Kafka消费者项目，它使用来自特定Kafka主题的数据。我一得到90%的记录就进行了处理，但我延迟了对其中一些记录的处理(10%)。
这些记录需要延迟，我无法提交记录，因此可能会导致Kafka将分区重新分配给新节点。为了避免这种情况，我可以将同一主题读两次，并在第二个使用者中延迟获取数据部分，但它需要两次反序列化，因此会带来开销。
有没有可能使用单个消费者读取记录，但与Kafka消费者有两个单独的提交？就提交而言，这基本上类似于有两个不同的消费者，consumer.poll将从单个消费者调用，但每个批处理将有两个consumer.commitSync。我将帮助我避免额外的反序列化和网
浏览 0提问于2017-12-27得票数 0
我想使用Kafka高级消费者API，同时我想禁用偏移量的自动提交。我试图通过以下步骤来实现这一点。
1) auto.commit.enable = false
2) offsets.storage = kafka
3) dual.commit.enabled = false
我创建了一个偏移管理器，它定期创建对kafka的offsetcommit请求并提交偏移。
我仍然有以下问题
1)高级消费者API是否会自动从kafka存储中获取偏移量，并使用该偏移量进行初始化？或者我应该使用简单的消费者API来实现这一点？
2)是否所有代理都支持基于kafka的偏移量存储？或者只在一个代理上维护它？
浏览 1提问于2015-04-28得票数 5
1回答
我的应用程序有使用web应用程序提交csv文件的用户。
kafka生产者读取每个CSV记录，并发送到具有2个分区和每个分区中的2个消费者的kafka主题。每个CSV文件可以有100 - 100,000条记录。
每个消费者阅读同一主题，平均30秒可处理1000条记录。
因此，4个消费者可以在( 100,000 /4)*30/1000/60 = 12.5分钟内处理一个100,000条记录的文件。
问题是，当一个文件正在处理时，第二个文件必须等待12.5分钟，第三个文件必须等待25分钟，以此类推。
如何管理我的分区策略，使每个文件都是并行处理而不是串行处理的？
此外，如果正在处理一个大文件，我是否可
浏览 22提问于2020-11-12得票数 0
1回答
我有一个春季卡夫卡消费者，它读取记录，并将它们交给一个缓存。计划的任务将定期清除缓存中的记录。我只想在批处理成功保存到数据库后才更新提交偏移量。我尝试将确认对象传递给缓存服务，以调用如下所示的应答方法。
public class KafkaConsumer {
@KafkaListener( topicPattern = "${kafka.topicpattern}", containerFactory = "kafkaListenerContainerFactory" )
public void receive( ConsumerRecord&
浏览 0提问于2019-01-26得票数 2
2回答
我有一个负载不是恒定的系统。我们可能一天收到1000个请求，或者根本没有请求。
我们使用Kafka在服务之间传递请求。我们保留了卡夫卡消费者的平均数量，以降低成本。现在，我的卡夫卡消费者将坐在理想的位置，如果没有收到当天收到的请求，如果收到太多的请求，就会有延迟。
我们希望将这些用户保持在自动模式下，这样，如果请求数量激增，我的服务器数量(Kafka使用者)就会增加。一旦请求数量减少，我们将删除服务器。因此，卡夫卡分区必须相应地增加或减少。
Kafka允许增加分区。在这种情况下，我们如何减少Kafka分区dynamically?Is，还有其他解决方案来处理这个自动缩放吗？
浏览 17提问于2021-02-09得票数 0
1回答
我有一个来自订阅主题的消费者投票。它会消耗每条消息，并进行一些处理(在几秒钟内)，推送到不同的主题并提交偏移量。
总共有5000条信息，
在重新启动之前使用2900条消息和提交的偏移量。
重新启动后-从偏移量0开始消耗。
即使使用相同的使用者组创建使用者，它也开始处理来自偏移量0的消息。
kafka版本(strimzi) > 2.0.0 kafka-python == 2.0.1
浏览 1提问于2020-10-02得票数 1
1回答
我正在考虑使用Kafka作为一个分布式工作队列，多个工作人员可以从其中检索任务。我原来的设计如下：
Work Producer ---> Kafka topic ------worker 1
|
|__worker 2
...
|__worker n
这个设计的问题是：
如果某个工作人员从主题中提取任务并立即提交偏移量，
浏览 2提问于2020-09-30得票数 3
回答已采纳
2回答
我正在为一个简单的数据管道实现只需一次的语义，使用Kafka作为消息代理。通过设置set enable.idempotence=true，我可以强制Kafka producer对每条生产记录只写一次。
然而，在消费方面，我需要保证消费者只读取每条记录一次(我对将消费的记录存储到外部系统或另一个正在处理的Kafka主题中不感兴趣)。要实现这一点，我必须确保处理轮询记录，并将它们的偏移量以原子/事务方式提交给__consumer_offsets主题(两者都成功/失败)。
在这种情况下，我是否需要求助于Kafka事务API来在消费者轮询循环中创建事务生产者，其中在事务中我执行：(1)处理所使用的记录
浏览 0提问于2020-10-10得票数 1
1回答
我有一个java客户端(使用kafka客户端版本2.4.1)，它连接到一个kafka (版本2.4.1)，其中有5个主题，每个主题5个分区。我正在为每个分区创建一个线程。我的问题是，对于每个线程，所有的消费者配置都会被记录下来： ? 首先，我想知道这是否是正常行为(因为我有25个线程，我看到的是这些日志的25倍)？也许我做错了什么。目前，我的每个使用者线程都在订阅它的主题。所以我调用了subscribe方法25次，每次调用似乎都会生成这些日志。 那我能做些什么来禁用它们吗？
浏览 6提问于2020-04-07得票数 1
回答已采纳
1回答
我有两个主题，每个主题上有10个分区，我使用下面的代码来收听消息。这里将建立多少个连接？是20吗？每个连接是在分区级别还是在命名空间(bootstarap服务器)级别？
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaListenerContainerFactory() {
final ConcurrentKafkaListenerContainerFactory<Object, Objec
浏览 0提问于2021-06-08得票数 1
1回答
我正在使用spring-kafka和消费卡夫卡主题的批量记录，并通过AbstractMessageListenerContainer.AckMode.BATCH提交偏移量。
在我的例子中，处理批处理记录需要时间(大约20秒)，使用者线程等待批处理过程完成，然后再次执行轮询(在此轮询时提交偏移量)。在本例中，我将记录的List分配给一个线程(名称：ProcessThread)，该线程将处理所有记录并将结果返回给使用者线程，然后使用者线程将记录结果。(在此过程中，使用者线程将一直等待，直到它从ProcessThread获得结果，这会导致性能低下。
有没有办法让ProcessThread负责将off
浏览 6提问于2018-10-03得票数 1
所以基本上我看到的是，我正在使用@KafkaListener消费消息，我故意不想承认它们，所以消费者忽略它们，并在主题中侦听新的消息。因此，如果应用程序重新启动，使用者只会从最后提交的消息开始。因此，我希望避免这种情况，我希望消费者在不重新启动的情况下考虑未提交的消息。下面是我正在使用的配置和@kafkalistener。
ReceiverConfig.java
package com.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serializ
浏览 3提问于2019-12-05得票数 1
我们有不同的消费者组，他们使用来自不同topics.We的数据，topic.We的不同分区需要允许一些消费者组访问特定的主题，以便只有这些组可以从主题中读取。我尝试了像这样的融合Kafka访问控制列表：
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Bob --consumer --topic test-topic --co
浏览 9提问于2017-07-12得票数 1
使用该配置
使用spring-integration-kafka扩展和以下配置：
<int-kafka:zookeeper-connect id="zookeeperConnect"
zk-connect="#{kafkaConfig['zooKeeperUrl']}" zk-connection-timeout="10000"
zk-session-timeout="10000" zk-sync-time="2000" />
<int-kafka:consume
浏览 0提问于2016-01-29得票数 0
1回答
我可以使用KafkaConsumer在单独的线程中使用消息。
但是，当我使用multiprocessing.Process而不是threading.Thread时，我会得到一个错误：
OSError: [Errno 9] Bad file descriptor
这个和建议使用多处理并行地使用消息是可能的。有谁能给我举个例子吗？
编辑
这是一些示例代码。不好意思，原来的代码太复杂了，所以我在这里创建了一个示例，希望能够告诉您正在发生的事情。如果我使用threading.Thread而不是multiprocessing.Process，这段代码可以正常工作。
from multiprocessin
浏览 3提问于2017-09-29得票数 4
回答已采纳
我们有一个HDP集群，有3个kafka经纪人(来自hortonworks )。
我们希望运行kafka控制台使用者，以便从带有特定偏移量的主题中获得一条消息。
/usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh --zookeeper zoo01:2181 --topic lopet.lo.pm--partition 0 --offset 34537263 --max-messages 1
但我们得到了以下信息：
我们哪里错了？
Using the ConsoleConsumer with old consumer is d
浏览 3提问于2019-12-11得票数 5
回答已采纳
1回答
在分区0上测试一条Kafka消息，处理时间超过一个小时。我尝试过自动提交，暂停/恢复，增加最大轮询超时，但我无法停止消费者消息复制。这是使用python的融合式kafka。我看不出有什么方法可以实验异步/同步提交。
msg= self.consumer.poll(timeout=1.0)
if msg:
self.consumer.pause()
some task.. (takes ~60 minutes minutes)
self.consumer.res
浏览 3提问于2022-03-07得票数 0
2回答
我们有一个应用程序，消费者读取消息，线程执行许多操作，包括在消息被生成到另一个主题之前访问数据库。在线程上消费消息和生成消息之间的时间可能需要几分钟。一旦向新主题生成了消息，就会执行一次提交，以指示我们已经完成了对消费者队列消息的处理。由于这个原因，自动提交被禁用。
我使用的是高级消费者，我注意到zookeeper和kafka会话超时，因为我们在消费者队列上做任何事情之前花费的时间太长了，所以kafka最终在每次线程返回以从消费者队列读取更多信息时重新平衡，并且在一段时间后消费者读取新消息之前开始花费很长时间。
我可以将zookeeper会话超时设置得非常高，以避免造成问题，但随后我必须相应地
浏览 2提问于2014-12-20得票数 9
回答已采纳
在kafka连接分布式模式下，需要通过REST提交kafka连接源和kafka连接接收器配置。卡夫卡如何利用卡夫卡集团协调人组建消费集团？
浏览 5提问于2022-09-19得票数 0
我在无头模式下运行Kafka和KSQLDB服务器。在KSQLDB服务器上，我只部署了几个查询来进行以下实验：
CREATE STREAM pageviews_original (viewtime bigint, userid varchar, pageid varchar) WITH (kafka_topic='pageviews-ksql', PARTITIONS=1, REPLICAS=3, value_format='DELIMITED');
CREATE TABLE users_original (registertime BIGINT, gender
浏览 1提问于2020-05-20得票数 3
回答已采纳
我有一个应用程序下载特定的网页内容，从一个从1 Kafka产生的URL流-生产者。我已经创建了一个有5个分区的主题，有5个kafka-consumers。然而，网页下载的超时时间是60秒。当其中一个url正在下载时，服务器会假定该消息已丢失，并将数据重新发送给不同的消费者。
我试过了我提到的所有东西
和
但我每次都收到不同的错误。
是否可以在kafka中将特定的消费者与分区绑定？我在我的应用程序中使用kafka-python
浏览 4提问于2017-08-29得票数 4
回答已采纳
1回答
嗨，我正在致力于将kafka升级到.9，并将kafka消费者升级到与.9一起发布的新java消费者。在升级时，我正在使用现有的主题，步骤只是停止.8 kafka并启动指向相同log.dirs的.9 kafka，在消费者端，我使用相同的组名和主题名称，但新的消费者正在从主题中的起始位置再次消费消息。我已经在.8中提交了它们。我添加了auto.offset.reset =最早的。
是否知道为什么会发生这种情况，或者我们需要首先与.8消费者一起使用所有消息，并让延迟变为0。谢谢，桑尼
浏览 0提问于2016-05-05得票数 0
默认情况下，亚马逊网络服务的Apach Kafka托管流是否将enable.auto.commit设置为true？我没有出现在文档中，我想确保我的消费者正确地提交了偏移量。 或者MSK遵循Kafka过程引用的here，如果消费者设置了ConsumerGroup ID，那么MSK会默认自动设置enable.auto.commit = true？
浏览 46提问于2021-09-13得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
