为什么没有将所有分区分配给正在运行的单个KafkaConsumer？
Kafka是一个分布式流处理平台，它通过将数据分成多个分区并将其分布在多个服务器上来实现高吞吐量和容错性。KafkaConsumer是Kafka提供的一个用于消费数据的客户端。在Kafka中，将一个主题（topic）划分为多个分区（partition），每个分区可以在不同的服务器上进行复制以实现容错性。
为了保证高可用性和负载均衡，Kafka采用了分区的机制。每个分区只能由一个消费者（KafkaConsumer）进行消费，这样可以确保每个消费者只消费一部分数据，从而实现并行处理和提高吞吐量。如果将所有分区分配给正在运行的单个KafkaConsumer，会导致该消费者负责处理所有数据，可能会造成负载过重，影响消费性能。
另外，将所有分区分配给单个消费者还存在单点故障的风险。如果该消费者发生故障或停止运行，所有分区将无法被消费，导致数据丢失或无法及时处理。
因此，将所有分区分配给正在运行的单个KafkaConsumer是不推荐的做法。相反，建议使用多个消费者来共同消费分区，以实现负载均衡和高可用性。Kafka提供了一种称为消费者组（Consumer Group）的机制，可以将多个消费者组织在一起，每个消费者负责消费一个或多个分区。这样可以实现并行处理和提高整体的消费能力。
腾讯云提供的相关产品是TDMQ，它是一种高性能、低延迟、高可靠的消息队列服务，可以满足分布式消息通信和数据传输的需求。TDMQ支持类似Kafka的分区机制，可以将消息分发到多个消费者进行并行处理，提供了高吞吐量和高可用性的消息传输能力。您可以了解更多关于TDMQ的信息和产品介绍，可以访问腾讯云官网的TDMQ产品页面：https://cloud.tencent.com/product/tdmq
相关·内容
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
回答已采纳
1回答
我有一个应用程序，它有两个线程正在运行。这些线程在它们之间共享一个KafkaConsumer实例。
KafkaConsumer是通过禁用自动提交配置创建的。一个线程将根据某些外部输入将主题分区分配给KafkaConsumer实例。另一个线程不断轮询主题，处理记录并在结束时提交偏移量。
现在，假设主题-分区分配在线程-1中发生了更改。我们如何确保Thread-2能够看到这些更改？也就是说，线程-2的KafkaConsumer将知道主题-分区已经在线程-1中更新。
将变量定义为易失性是否解决了这个问题？还是我们应该再做点什么？请提供您的解释与代码示例，因为我是新的多线程在Java。
浏览 4提问于2021-10-26得票数 2
回答已采纳
我在AWS的DC/OS (Mesos)集群上安装了Kafka。启用了三个代理，并创建了一个名为"topic1“的主题。
dcos kafka topic create topic1 --partitions 3 --replication 3
然后，我编写了一个Producer类来发送消息，一个Consumer类来接收消息。
public class Producer {
public static void sendMessage(String msg) throws InterruptedException, ExecutionException {
Map
浏览 1提问于2017-02-28得票数 8
回答已采纳
我使用的是KafkaConsumer 0.10Java api。我想从一个特定的分区和特定的偏移量消费。我抬头一看，发现有一个seek方法，但它抛出了一个异常。有没有人有类似的用例或解决方案？
代码：
KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(consumerProps);
consumer.seek(new TopicPartition("mytopic", 1), 4);
异常
java.lang.IllegalStateException: No current ass
浏览 0提问于2016-12-07得票数 21
回答已采纳
1回答
我有一个设置，其中每个KafkaConsumers在一个主题上处理多个分区。它们被静态地分配分区，以确保每个使用者都有相同数量的分区要处理。还选择了记录键，以便在所有分区上具有相同的消息分布。
在负载较重的时候，我们经常会看到少数分区会产生相当大的延迟(数千条消息/几分钟值)，而其他正在获得相同负载且被同一个使用者占用的分区则设法将延迟降低到几百条消息/几秒钟。
看起来消费者正在尽可能快地获取记录，遍历大部分分区，但有时会有一个分区被遗漏很长时间。理想情况下，我希望看到延迟更均匀地分布在各个分区上。
我已经阅读了一段时间关于KafkaConsumer民意测验行为和配置的文章，到目前为止，我认为
浏览 5提问于2019-12-27得票数 5
1回答
卡夫卡试验容器片状试验
、、、、
把卡夫卡融入我的春季项目。已经使用TestContainer方法编写了一个集成，但是测试不时失败。似乎是卡夫卡服务器初始化的问题。
下面是我的代码
def setupSpec() {
kafka = new KafkaContainer()
kafka.start()
System.setProperty("kafka.consumer.endpoint", kafka.bootstrapServers.replace("PLAINTEXT://", ""))
}
def setup() {
RestAssur
浏览 0提问于2018-12-24得票数 0
回答已采纳
2回答
我无法理解这个API的设计！
在下面的代码中，我们订阅了一个具有动态分配分区的主题列表。这完全没问题。
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
consumer.subscribe(Arrays.asList("some-topic"));
while(true){
ConsumerRecords<String, String> records = consumer
浏览 0提问于2019-07-27得票数 2
我有一个KafkaConsumer并使用手动分区分配。为了分发分区，我定期使用consumer.partitionsFor(topicId)来检测添加的分区，因为作业将永远运行，我希望支持这种情况。但是，这总是返回分区的初始列表，除非我重新启动使用者。
是否有方法检测来自使用者的附加分区？投票或听什么？
浏览 2提问于2017-09-12得票数 0
回答已采纳
1回答
我正在实现Kafka消费者类来接收信息。我只想每次都收到新消息。因此，我将enable.auto.commit设置为true。然而，这种抵消似乎一点也不改变。尽管主题、使用者组和分区始终是一样的。
这是我的消费者代码：
consumerConfig.put("bootstrap.servers", bootstrap);
consumerConfig.put("group.id", KafkaTestConstants.KAFKA_GROUP);
consumerConfig.put("enable.auto.commit"
浏览 4提问于2017-03-09得票数 1
2回答
我正在尝试使用assign()方法消费来自多个主题的消息。通过我的实现，有时我能够使用来自所有主题的消息，而另一些时候只能使用一个主题的消息。经过研究，我发现Kafka默认使用范围赋值器。因此，它不会总是分配所有分区。
对于我的用例，我应该能够使用所有主题和分区。
我已经尝试通过设置RoundRobin分配器。但这并没有帮助
List<TopicPartition> topicPartitions = new ArrayList<>();
KafkaConsumer kafkaConsumer = new KafkaConsumer<>(consumerCo
浏览 0提问于2019-06-17得票数 0
我是卡夫卡的新手，我通过了，但我什么都不懂。有人能解释一下什么时候使用ConcurrentKafkaListenerContainerFactory类吗？我已经使用了Kafkaconsumer类，但是我发现在我当前的项目中正在使用ConcurrentKafkaListenerContainerFactory。请解释一下它的用途。
浏览 2提问于2019-03-06得票数 11
回答已采纳
当我试图接收来自主题的消息时，下面定义的kafka使用者con工作得非常好；但是，当我试图使用seek方法或它的任何变体更改偏移量时，它会给我带来麻烦。即seek_to_beginning，seek_to_end
from kafka import KafkaConsumer, TopicPartition
con = KafkaConsumer(my_topic, bootstrap_servers = my_bootstrapservers, group_id = my_groupid)
p = con.partitions_for_topic(my_topic)
my_partitio
浏览 1提问于2017-08-01得票数 4
回答已采纳
1回答
目标:读取主题中的所有消息，然后终止该过程。
我可以通过以下内容连续阅读消息：
props.put("bootstrap.servers", kafkaBootstrapSrv);
props.put("group.id", group_id);
props.put("max.poll.records", 1); // Only get one record at a time. I understand that to read all messages this will need to be increased
props.put("
浏览 16提问于2018-12-14得票数 3
回答已采纳
1回答
我试图让x数的消费者访问kafka中指定的主题，但不使用相同的消息。比如我想要..。
消费者1拾取偏移1消费者2拾取偏移2消费者1拾取偏移3消费者2拾取偏移4
我想让卡夫卡为这两位消费者排队。我注意到了group.id配置，我假设您可以使用同一个组，它会相应地处理它，但它似乎不像我想的那样工作。
这是我使用的代码..。
public void init(){
Properties props = new Properties();
props.put("bootstrap.servers", kafkaUrl);
浏览 2提问于2016-12-06得票数 0
回答已采纳
我试着让两个消费者订阅一个EmbeddedKafkaBroker。第一个成功了，第二个失败了。@EmbeddedKafka和@ClassRule brokers都失败了。
@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka(topics = { "topic" })
public class AnnotationEmbeddedKafkaTest {
@Autowired
private EmbeddedKafkaBroker broker;
@Test
public voi
浏览 17提问于2021-05-18得票数 0
1回答
kafka版本: 0.9.0.1
如果是n = 20，我必须获取一个主题的最后20条消息。
我试着用
kafkaConsumer.seekToBeginning();
但它会检索所有的消息。我只需要收到最后20条消息。
此主题可能有数十万条记录
public List<JSONObject> consumeMessages(String kafkaTopicName) {
KafkaConsumer<String, String> kafkaConsumer = null;
boolean flag = true;
List<JSONObject>
浏览 0提问于2019-04-18得票数 3
回答已采纳
1回答
我创建了一个包含3个分区的主题
Topic:fixtures PartitionCount:3 ReplicationFactor:1 Configs:
Topic: fixtures Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: fixtures Partition: 1 Leader: 0 Replicas: 0 Isr: 0
Topic: fixtures Partition: 2 Leader: 0 Replicas: 0 Isr: 0
我使用Java Producer同步
浏览 4提问于2018-02-27得票数 1
回答已采纳
2回答
我想使用Pyhon和Kafka客户端获取Kafka主题的承诺偏移量和结束偏移量(位置)，
我想知道为什么偏移量由topic + consumer-group本身(KafkaAdminClient.list_consumer_group_offsets(consumer_group))接收
而位置是由消费者+主题的分区(KafkaConsumer.end_offsets(partitions))接收的。
有谁有线索吗？
浏览 3提问于2020-06-07得票数 0
注意:使用kafka_2.11-0.9.0.1
我用3 Partitions创建了一个名为：consumer-tutorials的kafka主题，如下所示：
C:\kafka_2.11-0.9.0.1>.\bin\windows\kafka-topics.bat --describe --topic consumer-tutorial
--zookeeper localhost:2181
Topic:consumer-tutorial PartitionCount:3 ReplicationFactor:1 Configs:
Topic: consu
浏览 2提问于2016-06-21得票数 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
2回答
我有一个需要订阅两个主题topicA和topicB的KafkaConsumer。然而，我需要一些参数是不同的。例如，如果我需要topicA的auto.offset.reset是earilest，而对于topicB，它应该是latest。我看不出有什么简单的方法可以做到这一点。一种选择是运行两个消费者，但在这种情况下，我需要为它们提供两个轮询线程，因此应该处理多线程。有没有更简单的方法？
浏览 1提问于2021-07-13得票数 0
我有以下代码
class Consumer(val consumer: KafkaConsumer<String, ConsumerRecord<String>>) {
fun run() {
consumer.seekToEnd(emptyList())
val pollDuration = 30 // seconds
while (true) {
val records = consumer.poll(Duration.ofSeconds(pollDuration))
浏览 2提问于2021-12-03得票数 6
回答已采纳
我正在使用Kafka版本2.0和java使用者API来使用来自主题的消息。我们使用的是一个单节点Kafka服务器，每个分区只有一个使用者。我注意到消费者正在失去一些信息。场景是:消费者投票这个话题。我已经创建了一个消费者每线程。获取消息并将其交给处理程序来处理消息。然后用“至少一次”的卡夫卡消费者语义学来提交抵消，以实现卡夫卡的偏移。同时，我有另一个使用者使用不同的组-id运行。在这个使用者中，我只是增加消息计数器并提交偏移量。在这个消费者身上没有信息丢失。
try {
//kafkaConsumer.registerTopic();
consumerThread = new
浏览 1提问于2019-05-10得票数 1
这是两个部分的问题。
一、守则是：
import java.time.Duration
import java.util
import java.util.Properties
import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import scala.collection.Ja
浏览 6提问于2020-03-05得票数 1
回答已采纳
我有两个卡夫卡(2.11-0.11.0.1)经纪人。主题的默认复制因子设置为2。生产者只将数据写入零分区。
我已经安排了执行者定期运行任务。当它以每分钟少量的记录(每分钟100条)来消耗一个主题时，它的作品就像一种魅力。但是对于巨大的主题(每分钟10K)，方法投票不返回任何数据。
任务是：
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.Kafk
浏览 2提问于2019-03-04得票数 1
回答已采纳
1回答
假设我在一个2 Kafka分区中有一个主题(Test)，而每个消费者组中的一个消费者组(X，Y)是消费主题。
现在我想知道同一分区中其他使用者组的偏移量。下面的伪代码将解释需要
*** Let's assume this is running in the context of consumer group X
TOPIC = “test”
// consumer for group x
Consumer<K, V> consumerX = new KafkaConsumer<>(consumerProperties);
consumerX.subscribe
浏览 0提问于2018-08-08得票数 1
我正在尝试从最早的偏移量中读取卡夫卡主题，然后通过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
2回答
我需要在多线程中使用Kafka分区的记录，每个线程上都有唯一的记录来处理。我有下面的代码，我不知道是什么错误
public class ConsumerThread implements Runnable {
public String name;
public ConsumerThread(String name){
this.name = name;
}
public Properties getDefaultProperty(){
Properties prop = new Properties();
pro
浏览 0提问于2018-10-14得票数 0
2回答
我已经看到了一个类似于的问题
但是，我仍然想知道来自特定分区的流数据是否不可能实现？我使用了卡夫卡消费策略在星火流订阅方法。
ConsumerStrategies.Subscribe
这是我尝试订阅主题和分区的代码片段，
val topics = Array("cdc-classic")
val topic="cdc-classic"
val partition=2;
val offsets=
Map(new TopicPartition(topic, partition) -> 2L)//I am not clear with this lin
浏览 0提问于2018-06-07得票数 0
回答已采纳
如何将python使用者脚本附加到特定的kafka分区。
在运行两个使用者脚本实例(如下所示)时，每个实例都随机获取一个分区，然后按照预期使用/打印该特定分区的所有消息。
但是，由于我需要将这些消息输出到磁盘上名为本地文件的分区，将脚本的每个实例附加到预先声明的分区ID将使事情变得更容易
档案名称(如。：
Date/Hour/PARTITION_ID-0.CSV
Date/Hour/PARTITION_ID-1.CSV
任何关于如何实现这一目标的想法。
请随意建议其他方法。
卡夫卡设置：
Topic:my-topic3 PartitionCount:2 ReplicationFactor
浏览 1提问于2016-06-25得票数 2
回答已采纳
我在Kubernetes上运行Kafka Connect (8-16个节点，自动标度)。我总共定义了44个连接器，每个Kafka主题一个(每个主题一个分区)。这些主题是由Debezium / Postgresql编写的。有3个Kafka节点。每个连接器的tasks.max设置为4。大部分连接器(但不是每个连接器！)有一个(总是一个)失败的任务，由于java.lang.IllegalStateException:没有当前分配的分区-0。
这里不是卡夫卡专家，请注意;)我假设有3个Kafka节点，因此3个工作人员做得很好，而第四个任务没有任何连接，所以失败了。但是为什么有时有4个任务运行得很好呢？
浏览 3提问于2019-03-24得票数 0
回答已采纳
1回答
我想使用成千上万的。例如，我需要一些体系结构模式的100_000使用者。我在想，可以吗？或者我应该重构我的系统，并在整个系统中使用很少的使用者(例如，10使用者而不是100_000)。
所以，我的问题是：
connection pool是否存在于KafkaConsumer中，还是每个消费者都与卡夫卡经纪人建立了自己的联系？
是否有thread pool在KafkaConsumer中，或者每个使用者创建自己的线程(我希望它没有)。
每个KafkaConsumer的平均内存消耗是多少？
你对这种建筑模式有何看法？
浏览 1提问于2019-02-27得票数 3
回答已采纳
我正在查看KafkaConsumer源代码。从源代码(我不太熟悉的java中)来看，暂停一个主题分区似乎将其偏移状态设置为暂停。我正在试图弄清楚KafkaConsumer在后台所做的取取发生了什么，以及暂停主题分区是否停止了kafka使用者所做的预取，或者它是否继续预取，但它在轮询中不返回该主题分区的ConsumerRecords。
浏览 2提问于2018-10-22得票数 0
回答已采纳
我从Apache开始，它是一个简单的生产者，Java的消费者应用程序。我正在使用kafka-clients版本的0.10.0.1并在Mac上运行它。
我创建了一个名为replicated_topic_partitioned的主题，它有3个分区，复制因子为3。
我在2181港开了动物园管理员。我在端口9092、9093和9094分别启动了三个id为1、2和3的代理。
以下是describe命令的输出
kafka_2.12-2.3.0/bin/kafka-topics.sh --describe --topic replicated_topic_partitioned --bootstrap-se
浏览 1提问于2019-08-21得票数 0
我正在编写一个概念应用程序的示例，以使用Apache 0.9.0.0中的消息，并查看是否可以使用它而不是普通的JMS消息代理，因为Kafka提供了一些好处。这是我的基本代码，使用新的使用者API：
public class Main implements Runnable {
public static final long DEFAULT_POLL_TIME = 300;
public static final String DEFAULT_GROUP_ID = "ltmjTest";
volatile boolean keepRunning =
浏览 2提问于2016-05-10得票数 3
回答已采纳
我试着从卡夫卡那里按分区消费。我正在使用KafkaConsumer api来完成这个任务。我看到这里提供了一个pause和resume方法，我可以从文档中收集到这是我想要的行为。
挂起从请求的分区获取。将来对轮询(Long)的调用将不会从这些分区返回任何记录，直到使用简历(Collection)恢复这些记录为止。请注意，此方法不影响分区订阅。特别是，当使用自动分配时，它不会导致组重新平衡。
我关心的是，我不会间歇地暂停和恢复KafkaConsumer，而是暂停所有的分区，除了我想要使用的分区。我担心暂停的调用代价高昂，会导致吞吐量问题。
我不确定这是否是实现来自KafkaConsume
浏览 0提问于2018-08-02得票数 1
我有一个有两个分区的kafka主题，我想创建一个使用者从给定的偏移量读取主题，下面是我用来从给定偏移量(9)读取的示例代码。
Properties configProperties = new Properties();
configProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
configProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.seri
浏览 0提问于2019-06-20得票数 1
回答已采纳
我有一个为特定主题(例如topic1)调用subscribe方法的KafkaConsumer。在订阅来自这个topic1的消息之后，我调用另一个主题的subscribe方法(例如topic2)。
最后，我调用了commitSync -是否会同时提交topic1和topic2偏移量？
非常感谢Stackoverflow-ers :)
浏览 2提问于2018-12-05得票数 0
我有一个玩具Flink作业，从三个卡夫卡主题，然后结合所有这三个流。仅此而已，没有额外的工作。
如果在我的Flink作业中使用并行性1，一切似乎都很好，因为当我更改并行性>1时，它失败了：
java.lang.OutOfMemoryError: Direct buffer memory
at java.nio.Bits.reserveMemory(Bits.java:693)
at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:123)
at java.nio.ByteBuffer.all
浏览 0提问于2019-03-07得票数 1
回答已采纳
我使用下面的代码来读取Kafka主题，并处理数据。
JavaDStream<Row> transformedMessages = messages.flatMap(record -> processData(record))
.transform(new Function<JavaRDD<Row>, JavaRDD<Row>>() {
//JavaRDD<Row> records = ss.emptyDataFrame().toJavaRDD();
浏览 1提问于2017-06-14得票数 11
2回答
我正在使用卡夫卡消费者阅读从几个主题，我需要其中一个有更高的优先权。处理过程需要很长时间，在(低优先级)主题中总是有很多消息，但是我需要尽快处理来自其他主题的消息。
这与类似，但这个问题使用的是旧的API。
在新API (0.10.1.1)中，有一些方法
KafkaConsumer::pause(Collection)
KafkaConsumer::resume(Collection)
但是，我不清楚如何有效地检测到高优先级主题中是否有新的消息，并且有必要暂停其他主题的消费。
有什么想法/例子吗？
浏览 0提问于2017-07-17得票数 3
回答已采纳
编辑
如果其他人在这种特殊的情况下，我得到了类似于我在调整消费者配置后寻找的东西。我创建了一个生产者，它将优先级消息发送到三个不同的主题(针对高/高/低优先级)，然后我创建了三个独立的消费者来消费。然后，我频繁地对优先级较高的主题进行调查，除非高优先级的主题是空的，否则不会对较低的优先级进行投票：
while(true) {
final KafkaConsumer<String,String> highPriConsumer = createConsumer(TOPIC1);
final KafkaConsumer<String,Str
浏览 0提问于2019-03-27得票数 5
流水线的简单代码如下：
source = env.addSource(kafkaConsumer)
.map(func).setParallelism(2).sink()
如何确定出库的顺序？
浏览 2提问于2018-08-27得票数 0
回答已采纳
2回答
要多次订阅给定的Kafka主题并从每个KafkaConsumer的每个分区接收每条消息，规范的方法是什么？
我目前正在做的是生成一个随机的Uuid group.id，这样每个订阅都是一个新的组，但是考虑到这些订阅是短暂的(并且有很多)，Kafka存储关于它们的元数据的开销可能是有害的。
实现这一点的正确方法是什么？
浏览 1提问于2020-09-07得票数 0
我使用的是kafka消费者api 0.10.2.1。
KafkaConsumer为分区分配和撤消提供了一个回调：
consumer.subscribe(topics, consumerRebalanceListener);
其中consumerRebalanceListener有两个方法：
public void onPartitionsRevoked(Collection<TopicPartition> partitions);
public void onPartitionsAssigned(Collection<TopicPartition> partitions
浏览 6提问于2017-10-18得票数 2
回答已采纳
以前我一直在使用0.8API。当您将主题列表传递给它时，它会返回一个流映射(每个主题一个条目)。这允许我生成一个单独的线程，并将每个主题的流分配给它。在每个主题中有太多的数据，产生一个单独的线程有助于多任务处理。
//0.8 code sample
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap =
consumer.createMessageStreams(topicCountMap);
我想升级到0.10。我检查了KafkaStreams和KafkaConsumer类。KafkaC
浏览 4提问于2016-09-30得票数 2
回答已采纳
用例：我们有为多模式配置的金门卡夫卡大数据适配器。每个模式映射到一个分区，kafka分区器按照定义的映射发布消息。比如schema1=1(Partition#1)，schema2=2 .scheman=n
这样，所有模式的消息都隔离在它们自己的分区中。
问题:给定情况下的，我想配置一个Apachi Nifi节点实例，以处理kafka主题的1个分区。这样，我将配置运行在并行处理100个kafka分区中的100个Nifi实例。
但我看不到在KafkaConsumer中可以指定分区的任何地方
问题：
Apache使用者支持分区配置吗？
如果它不支持，它是否在未来发布的路线图中？
浏览 2提问于2017-11-10得票数 1
回答已采纳
1回答
例如，如果在我的KafkaConsumer中有一个长时间运行的进程。(比方说要花一个小时来完成。)如果触发了再平衡，是否要等到重新平衡之前已开始处理的使用者使用(进程)消息时才对该使用者进行撤消操作？
那么，仅仅因为这种再平衡需要花费太多的时间，是否有可能？
还是立即执行撤销操作？
浏览 0提问于2019-03-12得票数 1
回答已采纳
我有一个关于2个代理的3个分区的主题。(Kafka版本: 0.8.1)
使用不同的用户guid (例如: FC42B34DD7658503E040970A2C437358)作为分区键批量添加消息。(约10k条消息)
加载消息时，我有一个正在运行的使用者(consumer1)，它开始很好地处理消息。
然后，我使用相同的消费者组id启动了另一个消费者(consumer2)。
我注意到的是，consumer1停止处理消息，而consumer2开始处理所有消息。
当我停止consumer2时，consumer1接管并恢复消息处理。
我希望两个消费者都能分担负载。
有什么线索可能是问题所在吗？谢谢。
浏览 0提问于2015-03-28得票数 1
对于下面的代码，我使用了
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
consumer.subscribe(Arrays.asList(TOPIC));
TopicPartition partition = new TopicPartition(TOPIC, 0);
System.out.println(consumer.position(partition));
我收到类似“您只能检查分配给此使用者的分区的位置”之类的错误，上面的代码有什么
浏览 0提问于2017-09-24得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
