建站
咨詢(xún)
售后
建站咨詢(xún)
新聞中心
消費(fèi)群是多線(xiàn)程或多機(jī)器的Apache Kafka主題。
創(chuàng)新互聯(lián)建站長(zhǎng)期為上千家客戶(hù)提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊(duì)從業(yè)經(jīng)驗(yàn)10年，關(guān)注不同地域、不同群體，并針對(duì)不同對(duì)象提供差異化的產(chǎn)品和服務(wù)；打造開(kāi)放共贏平臺(tái)，與合作伙伴共同營(yíng)造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為鎮(zhèn)海企業(yè)提供專(zhuān)業(yè)的成都做網(wǎng)站、成都網(wǎng)站設(shè)計(jì)，鎮(zhèn)海網(wǎng)站改版等技術(shù)服務(wù)。擁有10余年豐富建站經(jīng)驗(yàn)和眾多成功案例,為您定制開(kāi)發(fā)。
消費(fèi)者群體
消費(fèi)者可以使用相同的
group.id
加入群組一個(gè)組的最大并行度是組中的消費(fèi)者數(shù)量←不是分區(qū)。
Kafka將主題的分區(qū)分配給組中的使用者，以便每個(gè)分區(qū)僅由組中的一個(gè)使用者使用。
Kafka保證消息只能被組中的一個(gè)消費(fèi)者讀取。
消費(fèi)者可以按照消息存儲(chǔ)在日志中的順序查看消息。
重新平衡消費(fèi)者
添加更多進(jìn)程/線(xiàn)程將導(dǎo)致Kafka重新平衡。 如果任何消費(fèi)者或代理無(wú)法向ZooKeeper發(fā)送心跳，則可以通過(guò)Kafka集群重新配置。 在此重新平衡期間，Kafka將分配可用分區(qū)到可用線(xiàn)程，可能將分區(qū)移動(dòng)到另一個(gè)進(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客戶(hù)端API主配置設(shè)置的配置設(shè)置如下所示
| S.No | 設(shè)置和說(shuō)明 |
|---|---|
| 1 | 引導(dǎo)代理列表。 |
| 2 | group.id 將單個(gè)消費(fèi)者分配給組。 |
| 3 | enable.auto.commit 如果值為true，則為偏移啟用自動(dòng)落實(shí)，否則不提交。 |
| 4 | auto.commit.interval.ms 返回更新的消耗偏移量寫(xiě)入ZooKeeper的頻率。 |
| 5 | session.timeout.ms 表示Kafka在放棄和繼續(xù)消費(fèi)消息之前等待ZooKeeper響應(yīng)請(qǐ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)閮蓚€(gè)消費(fèi)者創(chuàng)建了一個(gè)示例組名稱(chēng)為 my-group
。 同樣，您可以在組中創(chuàng)建您的組和消費(fèi)者數(shù)量。
輸入
打開(kāi)生產(chǎn)者CLI并發(fā)送一些消息 -
Test consumer group 01 Test consumer group 02
第一個(gè)過(guò)程的輸出
Subscribed to topic Hello-kafka offset = 3, key = null, value = Test consumer group 01
第二個(gè)過(guò)程的輸出
Subscribed to topic Hello-kafka offset = 3, key = null, value = Test consumer group 02
希望你能通過(guò)使用Java客戶(hù)端演示了解SimpleConsumer和ConsumerGroup。
現(xiàn)在，您了解如何使用Java客戶(hù)端發(fā)送和接收消息。 讓我們?cè)谙乱徽吕^續(xù)Kafka與大數(shù)據(jù)技術(shù)的集成。
文章題目：創(chuàng)新互聯(lián)Apache Kafka教程：Apache Kafka 消費(fèi)者組示例
路徑分享：http://www.dlmjj.cn/article/cogccce.html
