建站
咨詢
售后
建站咨詢
新聞中心
消費(fèi)群是多線程或多機(jī)器的Apache Kafka主題。
創(chuàng)新互聯(lián)建站長期為上千家客戶提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊從業(yè)經(jīng)驗10年，關(guān)注不同地域、不同群體，并針對不同對象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺，與合作伙伴共同營造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為鎮(zhèn)海企業(yè)提供專業(yè)的成都做網(wǎng)站、成都網(wǎng)站設(shè)計，鎮(zhèn)海網(wǎng)站改版等技術(shù)服務(wù)。擁有10余年豐富建站經(jīng)驗和眾多成功案例,為您定制開發(fā)。
消費(fèi)者群體
消費(fèi)者可以使用相同的
group.id
加入群組一個組的最大并行度是組中的消費(fèi)者數(shù)量←不是分區(qū)。
Kafka將主題的分區(qū)分配給組中的使用者，以便每個分區(qū)僅由組中的一個使用者使用。
Kafka保證消息只能被組中的一個消費(fèi)者讀取。
消費(fèi)者可以按照消息存儲在日志中的順序查看消息。
重新平衡消費(fèi)者
添加更多進(jìn)程/線程將導(dǎo)致Kafka重新平衡。 如果任何消費(fèi)者或代理無法向ZooKeeper發(fā)送心跳，則可以通過Kafka集群重新配置。 在此重新平衡期間，Kafka將分配可用分區(qū)到可用線程，可能將分區(qū)移動到另一個進(jìn)程。
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;
public class ConsumerGroup {
public static void main(String[] args) throws Exception {
if(args.length < 2){
System.out.println("Usage: consumer ");
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("key.deserializer",
"org.apache.kafka.common.serializa-tion.StringDeserializer");
props.put("value.deserializer",
"org.apache.kafka.common.serializa-tion.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer(props);
consumer.subscribe(Arrays.asList(topic));
System.out.println("Subscribed to topic " + topic);
int i = 0;
while (true) {
ConsumerRecords records = con-sumer.poll(100);
for (ConsumerRecord record : records)
System.out.printf("offset = %d, key = %s, value = %s\n",
record.offset(), record.key(), record.value());
}
}
} Consumer客戶端API主配置設(shè)置的配置設(shè)置如下所示
| S.No | 設(shè)置和說明 |
|---|---|
| 1 | 引導(dǎo)代理列表。 |
| 2 | group.id 將單個消費(fèi)者分配給組。 |
| 3 | enable.auto.commit 如果值為true，則為偏移啟用自動落實，否則不提交。 |
| 4 | auto.commit.interval.ms 返回更新的消耗偏移量寫入ZooKeeper的頻率。 |
| 5 | session.timeout.ms 表示Kafka在放棄和繼續(xù)消費(fèi)消息之前等待ZooKeeper響應(yīng)請求(讀取或?qū)懭?多少毫秒。 |
編譯
javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java
執(zhí)行
>>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":. ConsumerGroupmy-group >>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":. ConsumerGroup my-group
在這里，我們?yōu)閮蓚€消費(fèi)者創(chuàng)建了一個示例組名稱為 my-group
。 同樣，您可以在組中創(chuàng)建您的組和消費(fèi)者數(shù)量。
輸入
打開生產(chǎn)者CLI并發(fā)送一些消息 -
Test consumer group 01 Test consumer group 02
第一個過程的輸出
Subscribed to topic Hello-kafka offset = 3, key = null, value = Test consumer group 01
第二個過程的輸出
Subscribed to topic Hello-kafka offset = 3, key = null, value = Test consumer group 02
希望你能通過使用Java客戶端演示了解SimpleConsumer和ConsumerGroup。
現(xiàn)在，您了解如何使用Java客戶端發(fā)送和接收消息。 讓我們在下一章繼續(xù)Kafka與大數(shù)據(jù)技術(shù)的集成。
網(wǎng)站標(biāo)題：創(chuàng)新互聯(lián)Apache Kafka教程：Apache Kafka 消費(fèi)者組示例
當(dāng)前URL：http://m.fisionsoft.com.cn/article/cogccce.html
