使用場(chǎng)景

大數(shù)據(jù)量、低并發(fā)、高可用、訂閱消費(fèi)場(chǎng)景

概念理解

分區(qū)個(gè)數(shù)與消費(fèi)者個(gè)數(shù)

分區(qū)個(gè)數(shù) = 消費(fèi)者個(gè)數(shù) ：最合適狀態(tài)

分區(qū)個(gè)數(shù) > 消費(fèi)者個(gè)數(shù) ：某些消費(fèi)者要承擔(dān)更多的分區(qū)數(shù)據(jù)消費(fèi)

分區(qū)個(gè)數(shù) < 消費(fèi)者個(gè)數(shù)? ：浪費(fèi)資源

當(dāng)“某些消費(fèi)者要承擔(dān)更多的分區(qū)數(shù)據(jù)消費(fèi)”，消費(fèi)者接收的數(shù)據(jù)不能保證全局有序性，但能保證同一分區(qū)的數(shù)據(jù)是有序的

groupId作用

采用同一groupId，分區(qū)個(gè)數(shù) >= 消費(fèi)者個(gè)數(shù)，每個(gè)消費(fèi)者都會(huì)消費(fèi)數(shù)據(jù)

采用同一groupId，分區(qū)個(gè)數(shù)<消費(fèi)者個(gè)數(shù)，某些消費(fèi)者不會(huì)接收數(shù)據(jù)

采用不同groupId，各個(gè)groupId的消費(fèi)者相互不受影響

命令行使用

啟動(dòng)：.\bin\windows\kafka-server-start.bat .\config\server.properties
創(chuàng)建topic：.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic lilei
開啟生產(chǎn)者：kafka-console-producer.bat --broker-list localhost:9092 --topic lilei
開啟消費(fèi)者：kafka-console-consumer.bat --zookeeper localhost:2181 --topic lilei

java api使用

api 包

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>1.0.0</version> </dependency>

生產(chǎn)者

package com.lilei.kafka.liei_kafka;import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; public class KafkaProducer { private final Producer<String, String> producer; public final static String TOPIC = "topic3"; private KafkaProducer() { Properties props = new Properties(); // 此處配置的是kafka的端口 props.put("metadata.broker.list", "127.0.0.1:9092"); props.put("zk.connect", "127.0.0.1:2181"); // 配置value的序列化類 props.put("serializer.class", "kafka.serializer.StringEncoder"); // 配置key的序列化類 props.put("key.serializer.class", "kafka.serializer.StringEncoder"); props.put("request.required.acks", "-1"); producer = new Producer<String, String>(new ProducerConfig(props)); } void produce() { int messageNo = 0; final int COUNT = Integer.MAX_VALUE; while (messageNo < COUNT) { String key = String.valueOf(messageNo); try {Thread.sleep(300);} catch (InterruptedException e) {// TODO Auto-generated catch block e.printStackTrace();}String data = "hello kafka message " + key; producer.send(new KeyedMessage<String, String>(TOPIC, key, data)); System.out.println(data); messageNo++; } } public static void main(String[] args) { new KafkaProducer().produce(); } }

消費(fèi)者

package com.lilei.kafka.liei_kafka;import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties;import kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import kafka.message.MessageAndMetadata; import kafka.serializer.StringDecoder; import kafka.utils.VerifiableProperties;public class KafkaConsumer { private final ConsumerConnector consumer; private KafkaConsumer() { Properties props = new Properties(); // zookeeper 配置 props.put("zookeeper.connect", "localhost:2181"); // group 代表一個(gè)消費(fèi)組 props.put("group.id", "vvvxyzv"); // zk連接超時(shí) props.put("zookeeper.session.timeout.ms", "5000"); props.put("zookeeper.sync.time.ms", "10000"); props.put("rebalance.max.retries", "10"); props.put("rebalance.backoff.ms", "2000"); props.put("auto.commit.interval.ms", "1000"); props.put("auto.offset.reset", "smallest"); // 序列化類 props.put("serializer.class", "kafka.serializer.StringEncoder"); ConsumerConfig config = new ConsumerConfig(props); consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config); } void consume() { String topic = "topic3";Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); topicCountMap.put(topic, new Integer(1)); StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties()); StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties()); Map<String, List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder); KafkaStream<String, String> stream = consumerMap.get(topic).get(0); ConsumerIterator<String, String> it = stream.iterator(); while (it.hasNext()) {MessageAndMetadata<String,String> mam = it.next();System.out.println(mam.key()+"---"+mam.message());} // System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<" + it.next().message() + "<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<"); } public static void main(String[] args) { new KafkaConsumer().consume(); } }

?

注意點(diǎn)

使用的kafka api版本要注意，在不合適或者存在bug的狀態(tài)下，會(huì)報(bào)：?kafka.common.ConsumerRebalanceFailedException

監(jiān)控

java -cp KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk localhost:2181 --port 8086 --refresh 10.seconds --retain 2.days

?

轉(zhuǎn)載于:https://www.cnblogs.com/lilei2blog/p/8608511.html

總結(jié)

以上是生活随笔為你收集整理的kafka概念使用简介注意点的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。