多线程Kafka消费者没有并行处理所有分区
是因为Kafka消费者的默认行为是每个线程只处理一个分区。这种行为可以确保消息的顺序性，但也会导致消费者在处理多个分区时无法充分利用多线程的优势。
为了实现多线程并行处理所有分区，可以采取以下步骤：
- 创建多个消费者线程：根据分区的数量创建相应数量的消费者线程。每个线程负责消费一个或多个分区的消息。
- 分配分区给消费者线程：使用Kafka提供的API，将分区分配给不同的消费者线程。这样每个线程就可以独立地消费分配给它的分区。
- 并行处理消息：每个消费者线程在独立的线程中运行，并行地处理分配给它的分区的消息。这样可以充分利用多线程的优势，提高消费速度和效率。
- 线程同步：在多线程环境下，需要注意线程之间的同步。可以使用线程安全的数据结构或者加锁机制来确保线程间的数据一致性和并发安全性。
优势：
- 提高消费速度和效率：通过多线程并行处理所有分区，可以充分利用多核处理器的优势，提高消息的消费速度和处理效率。
- 实现负载均衡：将分区均匀地分配给不同的消费者线程，可以实现消费者之间的负载均衡，避免某个线程负载过重。
- 提高系统的可伸缩性：通过多线程并行处理，可以根据实际需求增加或减少消费者线程的数量，从而提高系统的可伸缩性。
应用场景：
- 高吞吐量的消息处理：当需要处理大量消息时，多线程并行处理可以提高消息的消费速度和处理效率。
- 实时数据处理：对于需要实时处理的数据流，多线程并行处理可以确保消息能够及时被处理，满足实时性要求。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CKafka：https://cloud.tencent.com/product/ckafka
- 腾讯云容器服务 TKE：https://cloud.tencent.com/product/tke
- 腾讯云云服务器 CVM：https://cloud.tencent.com/product/cvm
- 腾讯云云原生应用引擎 TKE Serverless：https://cloud.tencent.com/product/tke-serverless
相关·内容
2回答
我想知道是否有可能读取和处理多个消息由Kafka消费者(Java客户端)并行..I意味着使用多线程...我应该使用rxJava吗?？
1)这是一个好的方法吗? 2)而且根据我的理解，Kafka甚至将每个线程都视为consumer...please，如果我错了，请纠正我……
3)并希望使Java客户端作为守护服务运行在Linux中，这样它就可以连续运行，轮询Kafka的消息，读取和处理相同的..is这是一个很好的方法。
浏览 0提问于2016-02-24得票数 0
2回答
我有一个负载不是恒定的系统。我们可能一天收到1000个请求，或者根本没有请求。
我们使用Kafka在服务之间传递请求。我们保留了卡夫卡消费者的平均数量，以降低成本。现在，我的卡夫卡消费者将坐在理想的位置，如果没有收到当天收到的请求，如果收到太多的请求，就会有延迟。
我们希望将这些用户保持在自动模式下，这样，如果请求数量激增，我的服务器数量(Kafka使用者)就会增加。一旦请求数量减少，我们将删除服务器。因此，卡夫卡分区必须相应地增加或减少。
Kafka允许增加分区。在这种情况下，我们如何减少Kafka分区dynamically?Is，还有其他解决方案来处理这个自动缩放吗？
浏览 17提问于2021-02-09得票数 0
我有一个有多个主题的Kafka集群，我将为每个主题设置一个分区，所有这些主题将由一个运行3个Kafka消费者线程的EC2实例消耗(每个线程一个消费者)，属于同一个消费者组。
我还没有尝试过，但我想知道Kafka是否可以平衡所有主题的分区，使其平均由3个线程消耗？或者Kafka会将所有分区都分配给一个线程使用？
浏览 1提问于2019-01-16得票数 1
1回答
我正在工作的火花流应用程序使用火花结构化流，汇合开源卡夫卡集群和运行星火作业在AWS EMR。我们至少有20+卡夫卡主题，将数据生成为雅夫罗格式的个别卡夫卡主题，每个主题被划分在3到4个分区之间。我正在使用SparkReadStream阅读所有20+主题(逗号分隔的主题值)。然后过滤结果DataFrame中的每个消息行，使用正确的Avro模式应用每条消息，并将得到的Dataset[T]写入S3和Cassandra。
有几个问题我找不到答案
对于所有的主题，我可以使用一个ReadStream吗？因为我在EMR中只执行一个“火花提交作业”，它会被认为是所有主题和分区的一个星火用户吗？
sp
浏览 2提问于2017-12-18得票数 4
如果我有一个应用程序发布关于kafka主题的事件，而我的消费者需要按照发布的顺序读取数据，那么我的主题只能有一个分区，因为kafka保证只在分区内排序。
然而，我读到kafka使用分区来提供可伸缩性，即在多个代理上放置一个主题的分区。我还读到，一个分区本身是不能分割的。
由于排序只能在分区内进行，所以对于我的应用程序来说，可伸缩性是一个问题吗？有没有办法解决这个问题，还是我对卡夫卡的理解不对？
假设我的应用程序有成千上万的使用者(每个人都在一个组中，所以每个人都使用发布的事件)。所有这些都需要使用单个分区从单个主题读取数据。
编辑:我脑海中的另一件事是:想象一下这个主题有5个分区，而所有消费者
浏览 6提问于2022-02-26得票数 1
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
2回答
我有一些基本的Kafka流代码，可以从一个主题读取记录，进行一些处理，并将记录输出到另一个主题。
Kafka streaming是如何处理并发的？所有的东西都在一个线程中运行吗？我在文档中没有看到这一点。
如果它是单线程的，我希望选择多线程处理来处理大量数据。
如果它是多线程的，我需要了解它是如何工作的，以及如何处理资源，比如SQL数据库连接应该在不同的处理线程中共享。
相对于其他选项(Spark、Akka、Samza、Storm等)，Kafka内置的流媒体API是否不推荐用于大流量场景？
浏览 1提问于2016-10-12得票数 27
回答已采纳
所有的人，请原谅我，我是一个新手刚刚开始的Kafka。目前，我正在阅读Kafka关于的文档。
如文件所述。
对于传统的信息系统。它们不能缩放消息处理。因为
发布-订阅允许将数据广播到多个进程，但由于每条消息都发送到每个订阅服务器，因此无法扩展处理。
我觉得这对我来说很有意义。
但是对于Kafka来说。文档说，即使在发布-订阅模式下，Kafka也可以扩展消息处理。(如果我错了，请纠正我。谢谢。)
卡夫卡的消费群体概念概括了这两个概念。与队列一样，使用者组允许您将处理划分为进程的集合(使用者组的成员)。与发布订阅一样，Kafka允许您向多个消费群体广播消息。
卡夫卡模型的优点在于，每个主
浏览 1提问于2017-07-16得票数 0
1回答
我正在开发一个使用Akka Kafka连接器的Kafka消费者应用程序。我希望消费者并行处理消息。我应该选择哪个消费群体？？如何在消费者端配置并行度？
浏览 16提问于2020-05-06得票数 0
我仍然不能清楚地理解并行性，比方说，我们有一个flink集群，它有足够的插槽。在我们的flink工作中，我们从3个不同的kafka集群中使用3个kafka主题，每个主题有10个分区。
如果我们想尽快使用该消息。并行性应该是什么？
如果我们将并行性设置为10，那么就会使用10个槽，这意味着，根据我的理解，10个线程对吗？如果这10个线程“连接”到topic_1，那么就没有线程读取topic_2和topic_3。
如果我们将并行性设置为30，那么就有30个线程，但这30个线程是否足够聪明，它们中的10个会转到topic_1，10个会转到topic_2，其余的10个会转到topic_3？
浏览 0提问于2019-03-13得票数 12
回答已采纳
2回答
我使用的是camel-kafka版本的2.14.3。下面是kafka URI： <from uri="kafka:{brokerlist}?topic={topic-name}&zookeeperHost={zookeeperHost}&zookeeperPort={zookeeperPort}&groupId={groupId-name}&consumerStreams=2" /> 请注意，我在URI选项中使用了consumerStream=2。但是，当我一次将多条消息发布到主题topic-name
浏览 85提问于2020-04-21得票数 2
回答已采纳
我正在尝试在Spark Structured和Hazelcast Jet之间映射一些概念，并理解其他主题。
Q1 -在Spark中，每个Kafka分区将成为spark中的一个分区，然后它们将由单个任务并行处理。我想我在某处读到过Hazelcast Jet将合并来自kafka的所有消息，而不考虑group.id和主题分区，对吗？
Q2 -我们如何在Jet程序中增加“消费者”的数量，以增加来自kafka的吞吐量消耗？在Spark中，我想我们只需要增加主题分区的数量，以便为新分区分配新的spark任务。
Q3 -如果上面的Q1是真的，有可能避免合并和分发要并行处理的kafka分区吗？一旦消息已经在k
浏览 0提问于2018-01-23得票数 2
我的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
使用apache-Kafka版本2.11-2.4.0。 我正在创建一个应用程序，我想在其中向同一消费者组中的所有订阅者(消费者)发布事件。但在许多地方，我发现Kafka向消费者群体广播消息。有没有办法做到这一点？
浏览 23提问于2020-01-09得票数 1
回答已采纳
2回答
Kafka分区延迟增加
、、、
我有一个使用Kafka 1.0作为队列的应用程序。Kafka主题有80个分区和80个消费者在运行。(Kafka-python消费者)。
通过运行以下命令：
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group mygroup --describe
我看到其中一个分区卡在偏移量上，随着新记录的添加，延迟不断增加。
上述命令的输出如下所示：
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
浏览 0提问于2017-11-15得票数 2
2回答
我正在使用KAFKA从一个来源获取数据。我想知道，如果我在微服务架构中串行增加组件数量，是否会使其更具可伸缩性？或者只是增加相同组件类型的KAFKA消费者的数量就会增加吗？对于任何错误，我很抱歉，我是第一次发布一个问题。
浏览 23提问于2020-04-27得票数 0
回答已采纳
在我的应用程序中，有多个企业。每个企业登录并执行一些操作，如上传数据，然后Kafka producer获取数据并发送到主题。另一方面，Kafka消费者消费来自主题的数据并执行业务逻辑。并持久保存到数据库中。在这种情况下，当单个企业登录时，一切都是完美的。但当多个企业登录时，Kafka将按顺序消费。也就是说，我怎样才能使进程并行？在多个客户端请求上。提前谢谢。
浏览 2提问于2019-02-12得票数 0
在Kafka文档中，我无法理解划分一个主题的确切需要是什么？如果它只是一个没有分区的主题(或者只有一个分区)，问题是什么？
浏览 8提问于2022-12-02得票数 0
根据Apache Kafka文档，消息的顺序可以在分区内实现，也可以在主题中的一个分区内实现。在这种情况下，我们得到的并行性好处是什么，它等同于传统的MQs，不是吗？
浏览 11提问于2015-04-23得票数 62
2回答
Kafka生产者和消费者延迟
、、、、
我的kafka生产者将在一秒内推送3条消息，我的消费者将消费，每条消息的数据处理时间为3秒。
所以在1分钟内生产者将产生180条消息，但是我的消费者在1分钟内只消费了20条消息。但是我想在1分钟内消费所有的消息
有没有人可以给我一个建议来解决这个问题，而不是增加实例
浏览 8提问于2021-07-13得票数 0
1回答
我一直在使用python模块来使用kafka代理。我想使用相同的主题并行地使用'x‘数量的分区。这些文件的内容如下：
# Use multiple consumers in parallel w/ 0.9 kafka brokers
# typically you would run each on a different server / process / CPU
consumer1 = KafkaConsumer('my-topic',
group_id='my-group',
浏览 2提问于2016-05-24得票数 8
回答已采纳
2回答
我正在使用spring来使用来自两个Kafka主题的消息，它发送的消息格式如下所示。
@KafkaListener(topics = {"topic_country1", "topic_country2"}, groupId = KafkaUtils.MESSAGE_GROUP)
public void onCustomerMessage(String message, Acknowledgment ack) throws Exception {
log.info("Message : {} is received
浏览 1提问于2020-05-24得票数 2
回答已采纳
2回答
Kafka Broker有3个分区。
卡夫卡消费者实例的计数是3。
突然，一个消费者的例子死了。
我知道，如果一个Kafka消费者实例死亡，Kafka代理正在进行再平衡，而另一个消费者实例将被分配到该分区。
我想知道这样的假设是否正确:另一个实例使用它最初使用的所有分区，然后分配和使用死分区。
(我是否必须在客户端代码中实现ConsumerRebalanceListener？)
如果是这样的话，在使用消息时会有什么延迟吗？
谢谢。
浏览 3提问于2017-05-17得票数 1
3回答
由于Kafka有一个基于主题的发布子体系结构，如何使用Kafka处理web应用程序的一对一和组消息部分？我正在使用SpringBoot+Angular堆栈和Docker服务器。
浏览 7提问于2020-05-09得票数 6
回答已采纳
我目前正在spark上运行一个流媒体应用程序，它使用新的直接方法(没有接收器)消费一个简单的kafka主题。由于我们在Kafka集群上遇到了一些问题，即使在Kafka上恢复正常后，流媒体应用程序也遇到了一些问题:一些作业正在处理几乎空的rdd，而另一些作业正在处理两倍大小的rdd。平均输入速率仍然与以前相同，但处理时间因此不规律。
我试着重启应用程序，但没有什么不同。我在Kafka和Spark两边都没有发现任何堆栈或线索。
有人能告诉我如何解决这类问题吗？或者至少到哪里去看一看，以了解发生了什么？
浏览 0提问于2016-05-18得票数 0
1回答
我正在尝试让多个不同的都连接到一个Kafka/Zookeeper集群，所有这些都在Kubernetes内部。微服务公司正在使用org.springframework.kafka:spring-kafka -作为事件的消费者和生产者。
所有的服务都连接到kafka，好的，主题是创建的，但是每个服务的使用者是非常不一致的。
例如，当服务一次启动时，所有使用者都将侦听消息并调用该函数。但是，当我重新启动一切(包括卡夫卡和动物园管理员)时，它要么不能工作，要么一些不同服务的消费者会工作等等。
下面是我的一些配置--我没有任何基于Java的配置--就在我的application.yml中，如下所示：
s
浏览 5提问于2020-11-28得票数 1
回答已采纳
2回答
多处理kafka消息
、、、
我们有一个kafka消费者，它不断地轮询kafka主题并处理消息。我们正面临着我们应该为max.poll.records赋予多少价值的问题。为了处理这条消息中的每条消息，我们有3个rest调用。我所做的是使用并行流以列表和处理的形式获取消息。实例有2 2GB内存空间，max.poll.records为8。因此，现在如果我增加max.poll.records，消息处理时间会增加，当max.poll.records为4时性能最佳。您能否说明我们如何使用1个实例使用1个分区来增加消息的并行处理，这意味着我可以尝试让它处理大约5条消息，而不是获取8条消息。
更新：-下面是内存配置，我正在pcf中部署独
浏览 5提问于2019-01-28得票数 1
我创建了3个Kafka经纪人设置与broker id的20,21,22。然后我创建了这个主题：
bin/kafka-topics.sh --zookeeper localhost:2181 \
--create --topic zeta --partitions 4 --replication-factor 3
其结果是：
当制片人向主题齐塔发送“你好世界”的信息时，卡夫卡首先将消息写到哪个分区？
在所有4个分区中复制"hello world“消息？
3个代理中的每个代理都包含所有4个分区？在上述情况下，这与3的复制因子有什么关系？
如果我有8个使用者在他们自己的进程或线
浏览 4提问于2020-07-30得票数 5
回答已采纳
1回答
我已经在kafka上工作了六个月了，我有一些关于消费者延迟和存储到主题中分区的数据的问题。
问题1:最初，当我开始阅读kafka并了解其功能如何使用时，我被反复传授一个包含一个部分和一个复制因子的主题将会带来奇迹。在整整六个月的工作之后，将我的项目转移到生活中的消费者，从主题消费我的消息开始给我一个滞后。我阅读了许多关于消费者延迟的堆栈溢出答案，并得出结论:如果我增加主题的分区和复制因子，就会删除它。我真正想知道的是，在六个月的数据流向某个主题后，这是否真的能消除我的消费者滞后，有人能帮助我在我的经纪人中消除这种滞后吗？提前谢谢。
问题2:举个例子，如果我增加了主题的部分和复制因子，那么从现在
浏览 20提问于2017-08-30得票数 0
回答已采纳
在我的设置中，我有一个包含三个进程(一个服务的3个实例)的使用者组，可以从Kafka消费。我发现发生的情况是，第一个节点正在接收所有流量。如果一个节点被手动终止，则下一个节点将获得所有Kafka流量，但最后一个剩余节点处于空闲状态。
所需的行为是所有消息均匀分布在消费者组内的所有实例中，这是我认为应该发生的情况。据我所知，Kafka的工作方式是，它应该在消费者组的所有成员之间均匀分发消息。我的理解正确吗？我一直在试图确定为什么消费者群体中只有一个成员没有运气地获得了所有流量。有什么想法/建议吗？
浏览 3提问于2014-06-26得票数 11
我正试图在ksqldb之上构建一个应用程序。
假设我会有一个简单的制作人：
package main
import (
"fmt"
"github.com/rmoff/ksqldb-go"
"net/http"
)
var client = ksqldb.NewClient("http://localhost:8088", "", "").Debug()
func init() {
offset := `SET 'auto.offset.reset
浏览 14提问于2022-02-14得票数 0
回答已采纳
这就是场景:我知道，使用与Spring相关的最新API (比如kafka 2.10)，我们可以这样做：
@KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic", partitions = { "0" }) })
@KafkaListener(id = "id1", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic",
浏览 0提问于2017-04-17得票数 1
4回答
我是Kafka的新手，我认为我在如何平衡分区队列的主题上遗漏了一些东西
我们在一个主题上有5个分区和2个消费者。该主题有一个空键，因此我假设Kafka以循环方式随机挑选一个新分区来添加新记录。
这意味着一个消费者将从3个分区中读取数据，而另一个消费者将从2个分区中读取数据。如果我的假设是正确的(即记录在多个分区中平均不受信任)，则具有3个分区的消费者将执行更多的工作(多1.5倍)。这可能会导致一个消费者什么都不做，而另一个消费者则继续努力工作。
我认为你应该为消费者提供一个均匀可分的分区数量。
我是不是遗漏了什么？
浏览 3提问于2018-08-01得票数 0
我在本地机器上设置了3个代理，我用5分区bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 5 --topic test-partitions创建了一个主题
然后来描述我的test-partitions主题，
bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test-partitions
这会导致
Topic: test-partitions Part
浏览 2提问于2020-05-11得票数 0
回答已采纳
我想就以下问题提出一些建议--我在我的应用程序中使用了一个Consumer.committableSource。在测试期间，我发现应用程序将耗尽给定的分区，直到它在切换到下一个分区之前使用最新的条目，而不是在Kafka主题的分区之间循环运行。对于我的应用程序来说，这并不理想，因为它关心把事件放在Kafka上的时间顺序。这种详尽的读取分区的方法就像在时间上来回来回地读一样。
有什么想法，我可以调整消费者，以支持循环的分区消费而不是？谢谢!
浏览 1提问于2019-10-14得票数 1
回答已采纳
2回答
我正在尝试理解如何动态地扩展消耗大量主题的应用程序(不幸的是，我不能减少它们的数量-通过设计，每个主题都是针对特定类型的数据的)。
我希望我的应用程序集群共享所有200+主题的负载。例如，当新的应用程序节点添加到集群中时，它应该“窃取”旧节点的一些主题订阅，这样负载就会再次均匀分布。
据我所知，Kafka partinions/consumer group有助于并行化一个主题，而不是在多个主题之间共享负载。
浏览 2提问于2017-11-26得票数 3
1回答
因此，我有一个消费者组在运行并侦听来自某个主题的消息。如何将更多的消费者实例添加到消费组中？如果这就是我收听消息的方式 let consumer = new kafka.ConsumerGroup(options, topicname);
consumer.on("message", function(message) {
//process message here
}); 我尝试创建消费者并在选项中包含相同的组名， new kafka.Consumer(
client,
[
{ topic: topicName}
浏览 33提问于2020-10-07得票数 1
1回答
如何使用Apache Kafka生成/消费延迟消息？似乎标准的Kafka (和Java kafka客户端)功能没有这个功能。我知道我可以用标准的等待/通知机制自己实现它，但它似乎不是很可靠，所以任何建议和好的实践都是值得感谢的。
找到了，但没用。正如我所看到的: Kafka基于从文件系统的顺序读取，并且只能用于读取主题，简单明了地保持消息顺序。我说的对吗？
浏览 3提问于2017-12-27得票数 10
回答已采纳
3回答
我正在研究Apache-kafka，有一些困惑。请帮助我理解以下场景。
我有一个主题，在一个Kafka集群中有5个分区和5个代理。我在Partition 1(假设是P1 )中保持消息顺序，.I希望将P1的消息广播给10个消费者。
所以我的问题是:这10个消费者如何与主题分区p1交互。
浏览 0提问于2018-08-06得票数 0
2回答
假设我有3个Kafka经纪人，一个动物园管理员，50个生产者，50个消费者和1个主题(testTopic1)。并且所有的消费者都订阅了testTopic1。现在，我将与50个生产者同时向同一主题(testTopic1)发送50条消息。现在我希望Kafka集群不会同时向消费者发送超过40条消息。剩下的10个将保留在队列中或放弃它。
也许这是Kafka中的负载均衡。我不知道我将如何做这项工作。我是新来卡夫卡的，请帮帮忙。
浏览 17提问于2019-07-14得票数 0
在我的场景中，有一个队列，其中许多资源将消息放入该队列，而许多消费者读取消息并执行特定的工作。
对于这个场景，我使用以下命令在Kafka中创建了一个主题
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1
现在，我开发了一个java类来使用它。
public class ConsumerGroupExample {
private final ConsumerConnector consumer;
private
浏览 2提问于2015-04-11得票数 0
我正在尝试设计一个使用Kafka的处理代理来并行处理来自Kafka主题的消息。
我想确保接近准确-每消息处理一次在整个消费者群体，虽然可以容忍-至少一次。
我发现文档在许多方面不明确，我必须知道以下几个具体问题：
如果一条消息被发布到某个主题，它是只存在于主题中的所有分区之间，还是在可能的多个分区上复制？我读过可以支持这两种可能性的声明。
“偏移”是每个分区还是每个消费者/消费者组/分区？
当我启动一个新的使用者时，它是查看作为一个整体的使用者组的偏移量还是它被分配的分区的偏移量？
如果我想增加新的用户，并且没有免费的分区(我相信每个分区最多只能有一个用户)，kafka会重新平
浏览 7提问于2017-05-17得票数 0
我有十个Kafka制作人，每个人都写一个主题的不同分区。 我不知道哪个更有效。 让一个消费者监听十个分区，还是让十个消费者监听不同的分区？
浏览 64提问于2019-03-21得票数 0
2回答
我想为几个主题创建一个kafka消费者。consumer的方法构造函数允许我在订阅中传递主题列表的参数，如下所示：
private Consumer createConsumer() {
Properties props = getConsumerProps();
Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);
ArrayList<String> topicMISL = new ArrayList<>();
for (String s:
浏览 0提问于2020-06-22得票数 2
我正在学习卡夫卡，在这里阅读导言一节
特别是关于消费者的那部分。导言的第二至最后一段内容如下：
卡夫卡做得更好。通过在主题中具有并行性--分区--的概念，Kafka能够为用户进程池提供排序保证和负载平衡。这是通过将主题中的分区分配给使用者组中的使用者来实现的，这样每个分区就会被组中的一个消费者使用。通过这样做，我们确保使用者是该分区的唯一读者，并按顺序使用数据。由于有许多分区，这仍然平衡了许多使用者实例的负载。但是，请注意，不能有比分区更多的使用者实例。
我的困惑源于最后一句，因为在作者描述两个消费者组和一个4分区主题的那一段的图像中，消费者实例比分区多！
而且，不能有比分区更多的使
浏览 2提问于2014-09-17得票数 102
回答已采纳
我使用的是1.5.4 spring boot，java 8和kafka 2.2.0版本。我有5个代理和4个不同的主题，每50个分区。
我有一个这样的系统：
我有两三百万张唱片。我出口这些记录的2500,2500部分上的每个卡夫卡消息。我的意思是1条kafka消息由2500条记录组成，但我使用的是相同的键，这意味着所有这些记录都会进入相同的分区。我使用紧凑的主题，也有2-3百万的记录，我使用ktable - kstream连接和收集另一条记录。(此记录具有不同的键，如Id，以便重新分区)在此连接操作之后，我返回到我的原始键进行重新分区。我的kafka streams代码监听这个主题，并执行重新分
浏览 0提问于2019-11-14得票数 0
我们在消费者内部存在性能问题。更准确地说，是指对消息做了什么。
通常，我们得到2000行的文件，拆分为6000个Kafka事件。
我们的主题X配置了10个分区。在这种情况下，我们可以在一个给定组中启动最多10个使用者，每个使用者将附加到一个单独的分区，直到重新平衡。顺便提一下，在我们的例子中，分区中的消息排序是强制的。
问题是，每条消息最多需要6秒才能被消费者处理。所以整个文件大约在1小时内被处理完。
我还知道Kafka是为可伸缩性而设计的，这种可伸缩性是由主题的分区数量来管理的。
(一个显而易见的解决方案是重构消费者的代码。这是有计划的，但我们开发团队的工作负载是这样的，而且这种重构可能在几
浏览 1提问于2019-03-06得票数 0
1回答
卡夫卡在多线程情况下的订货保证
、、、、
据我所知，卡夫卡生产者和消费者都必须在每个主题上使用一个线程-分区，如果我们想要按顺序写入/读取记录。我说得对吗?或者他们在这种情况下使用多线程？
浏览 11提问于2022-03-18得票数 0
回答已采纳
在使用Spark Structured Streaming和Kinesis Streams时，我在重新处理流中积累的数据时遇到了读取不平衡的情况(而不是从最新数据读取)。
下图显示了作为流的一部分的运动碎片的读取速度的差异。
这使得spark作业丢弃了许多事件，因为事件时间非常不同的事件会被混淆，而那些被认为是旧的事件会被丢弃。
最近，一位团队成员建议改用Kafka。我对Apache Kafka解决这个问题持怀疑态度，因为AFAIK修复我上面描述的读取不平衡的唯一方法是在消费者级别引入某种协调。这就是Apache Flink的kinesis连接器在重新处理Kinesis stream
浏览 1提问于2021-07-04得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
