我需要在多個線程中使用來自 Kafka 分區的記錄，每個線程上都有唯一的記錄進行處理。我有以下代碼，我不知道是什么錯誤public class ConsumerThread implements Runnable { public String name; public ConsumerThread(String name){ this.name = name; } public Properties getDefaultProperty(){ Properties prop = new Properties(); prop.setProperty("group.id", "4"); prop.put("enable.auto.commit", "false"); prop.put("auto.offset.reset", "earliest"); prop.setProperty("bootstrap.servers", "localhost:9092"); prop.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); prop.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); prop.setProperty("max.poll.records","150"); return prop; } public void run() { TopicPartition tp = new TopicPartition("my.topic", 0); KafkaConsumer consumer = new KafkaConsumer(getDefaultProperty()); ArrayList tpList = new ArrayList<TopicPartition>(); tpList.add(tp); consumer.assign(tpList); ConsumerRecords poll = consumer.poll(1000); Iterator it = poll.iterator(); consumer.commitAsync(); while(it.hasNext()){ ConsumerRecord cr = (ConsumerRecord) it.next(); System.out.println("From "+this.name+" : "+cr.value()); } consumer.close(); System.out.println("Thread Exiting "+this.name); }}結果From Thread1 : produced_0From Thread1 : produced_1From Thread1 : produced_2From Thread1 : produced_3...From Thread1 : produced_136From Thread2 : produced_0From Thread2 : produced_1From Thread2 : produced_2From Thread2 : produced_3...預期的 ：From Thread1 : produced_0From Thread1 : produced_1From Thread1 : produced_2From Thread1 : produced_3...From Thread1 : produced_136From Thread2 : produced_4From Thread2 : produced_5From Thread2 : produced_6From Thread2 : produced_137
2 回答
波斯汪
TA貢獻1811條經驗 獲得超4個贊
就像 Lior Chaga 在他的評論中所說的那樣，您正在手動將主題分區分配給您的消費者。這不是執行此操作的推薦方法。最重要的是，您的所有消費者似乎都在使用完全相同的 groupID。利用這種結構，有兩個線程消費，如果消費者的至少一個有一個特定的消息，沒有其他線程會得到一個。如果您希望所有的消費者線程都獲得自己的“一組”消息，而不會相互中斷，那么您需要給它們不同的group.ids。
要訂閱主題以便它為您處理自動重新平衡，然后消費，您應該執行以下操作（取自下面鏈接的 KafkaConsumer javadoc）：
consumer.subscribe(Arrays.asList("foo", "bar"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
Kafka 官方 javadocs 有更詳細的解釋：https ://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html
添加回答
舉報
0/150
提交
取消