卡夫卡如何为每个主题存储补偿？

在投票Kafka时，我使用subscribe()函数订阅了多个主题。现在，我想设置要从每个主题读取的偏移量，而不必在每个主题的seek()和poll()之后重新订阅。迭代地调用seek()以反映每个主题名称，之前查询数据实现结果？如何准确存储在Kafka中的偏移量？卡夫卡如何为每个主题存储补偿？

我有一个分区每个主题，只有一个消费者阅读所有主题。

来源

2017-08-15 Raunaq Kochar

KAFKA STORE如何补偿每个主题？

卡夫卡已经从zookeeper到kafka经纪商的抵消存储。理由如下：

动物园管理员是不是服务于高写入负载，如偏移更新的好方法，因为饲养员路线，每条路线虽然写的每一个节点，因此没有进行分区的能力或以其他方式规模写道。我们一直都知道这一点，但是因为我们已经依赖zk，所以选择这种实现方式作为一种“方便的婚姻”。

卡夫卡存储偏移提交一个主题中，当消费者提交的偏移，卡夫卡发布一个提交偏移消息发送到“提交日志”主题，并保持在内存中的结构，它映射组/主题/分区用于快速检索的最新偏移量。更多的设计信息可以在中找到。

现在，我想设置我想要从每个主题读取的偏移量，而不必在每个主题的seek（）和poll（）后重新订阅。

有一个关于kafka管理工具重置偏移量的新功能。

kafka-consumer-group.sh --bootstrap服务器127.0.0.1:9092 --group 你的消费群--reset-偏移 --to偏移1 - 全主题--execute

来源

2017-08-15 12:35:10 GuangshengZuo

其实这是节省了偏移

https://www.quora.com/What-is-the-actual-role-of-Zookeeper-in-Kafka-What-benefits-will-I-miss-out-on-if-I-don%E2%80%99t-use-Zookeeper-and-Kafka-together

然后，在消费，和特定GROUP_ID，您可以选择一个特定的主题，内容如下的动物园管理员：

的主题中的所有消息
所有新邮件

对于您可以使用属性：

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

，而不是 “最早的” 其他选项：

最早：自动复位偏移最早偏移
最新：自动将偏移量重置为最新偏移量
无：th如果没有发现以前的偏移量，则向消费者排除异常或消费者的组
其他：向用户抛出异常。

这是一个卡夫卡消费者的一个示例：

import java.util.Properties; 
import java.util.Arrays; 
import org.apache.kafka.clients.consumer.KafkaConsumer; 
import org.apache.kafka.clients.consumer.ConsumerRecords; 
import org.apache.kafka.clients.consumer.ConsumerRecord; 
import org.apache.kafka.clients.consumer.ConsumerConfig; 


public class ConsumerGroup { 
    public static void main(String[] args) throws Exception { 
     if(args.length < 2){ 
     System.out.println("Usage: consumer <topic> <groupname>"); 
     return; 
     } 

     String topic = args[0].toString(); 
     String group = args[1].toString(); 
     Properties props = new Properties(); 
     props.put("bootstrap.servers", "localhost:9092"); 
     props.put("group.id", group); 
     props.put("enable.auto.commit", "true"); 
     props.put("auto.commit.interval.ms", "1000"); 
     props.put("session.timeout.ms", "30000"); 
     props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 
     props.put("key.deserializer",   
     "org.apache.kafka.common.serializa-tion.StringDeserializer"); 
     props.put("value.deserializer", 
     "org.apache.kafka.common.serializa-tion.StringDeserializer"); 
     KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); 

     consumer.subscribe(Arrays.asList(topic)); 
     System.out.println("Subscribed to topic " + topic); 
     int i = 0; 

     while (true) { 
     ConsumerRecords<String, String> records = con-sumer.poll(100); 
      for (ConsumerRecord<String, String> record : records) 
       System.out.printf("offset = %d, key = %s, value = %s\n", 
       record.offset(), record.key(), record.value()); 
     }  
    } 
}

来源

2017-08-15 05:26:50

我只是想知道你是否可以做一个'寻找（）'设置偏移每个主题，在我的情况下，只有一个消费者订阅了多个主题。对于例如如果我想从topic1偏移量1，topic2偏移量2等中读取，我可以这样做，而不用仅使用seek（）来重新订阅。 –

好吧，我只是尽量让我的答案尽可能完整。据我所知，你不能修改偏移量，正如我在答案中所说的那样，该值是内部存储在Zookeeper中的。因此，您拥有的选项基本上是读取主题中的所有消息，或者是从上次阅读主题后添加的所有消息。顺便说一下，您使用的是哪种消费者，只是为了检查我是否可以找到关于它的更多信息？ –

如果我想一次又一次从相同的偏移量读取，我将禁用自动提交并始终将seek（）指定为所需的偏移量。我希望那些作品！什么样的消费者意味着什么？ KafkaConsumer –

卡夫卡

的偏移被存储在消费者侧。每个消费者将通常在动物园管理员中存储每个主题的偏移量。

来源

2017-08-15 05:27:10 Solo

因此，每个seek都会存储该特定主题分区对的偏移量，对不对？每次寻求电话后我都不需要投票，对吧？ –

@RaunaqKochar是 – Solo

这是最新的卡夫卡版本，请参阅广生's的答案 –

卡夫卡如何为每个主题存储补偿？

回答

相关问题