kafka概念使用简介注意点
使用场景
大数据量、低并发、高可用、订阅消费场景
概念理解
分区个数与消费者个数
分区个数 = 消费者个数 :最合适状态
分区个数 > 消费者个数 :某些消费者要承担更多的分区数据消费
分区个数 < 消费者个数 :浪费资源
当“某些消费者要承担更多的分区数据消费”,消费者接收的数据不能保证全局有序性,但能保证同一分区的数据是有序的
groupId作用
采用同一groupId,分区个数 >= 消费者个数,每个消费者都会消费数据
采用同一groupId,分区个数<消费者个数,某些消费者不会接收数据
采用不同groupId,各个groupId的消费者相互不受影响
命令行使用
启动:.\bin\windows\kafka-server-start.bat .\config\server.properties
创建topic:.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic lilei
开启生产者:kafka-console-producer.bat --broker-list localhost:9092 --topic lilei
开启消费者: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>
生产者
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(); } }
消费者
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 代表一个消费组 props.put("group.id", "vvvxyzv"); // zk连接超时 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(); } }
注意点
使用的kafka api版本要注意,在不合适或者存在bug的状态下,会报: kafka.common.ConsumerRebalanceFailedException
监控
java -cp KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk localhost:2181 --port 8086 --refresh 10.seconds --retain 2.days
转载于:https://www.cnblogs.com/lilei2blog/p/8608511.html
kafka概念使用简介注意点相关推荐
- Kafka 概念、单机搭建与使用
目录 Kafka 概念.单机搭建与使用 基本概念介绍 Topic Producer Consumer Kafka单机配置,一个Broker 环境: 配置zookeeper 配置Kafka 使用Kafk ...
- Kafka消息队列简介
Kafka消息队列简介 1 基本概念 Broker Kafka集群包含一个或多个服务器,这种服务器被称为broker Topic 每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic. ...
- kafka入门:简介、使用场景、设计原理、主要配置及集群搭建
为什么80%的码农都做不了架构师?>>> kafka入门:简介.使用场景.设计原理.主要配置及集群搭建(转) 问题导读: 1.zookeeper在kafka的作用是什么? 2. ...
- Kafka概念和基本架构概述
Kafka概念和基本架构概述 一.Kafka介绍 Kafka是最初由 Linkedin 公司开发,是一个 分布式.分区的.多副本的.多生产者.多订阅者,基于zookeeper协调的分布式日志系统(也可 ...
- ERP的概念、简介与发展历程
ERP的概念.简介与发展历程 ERP--Enterprise Resource Planning 企业资源计划系统,是指建立在信息技术基础上,以系统化的管理思想,为企业决策层及员工提供决策运行手段的管 ...
- kafka入门:简介、使用场景、设计原理、主要配置及集群搭
李克华 云计算高级群: 292870151 195907286 交流:Hadoop.NoSQL.分布式.lucene.solr.nutch kafka入门:简介.使用场景.设计原理.主要配置及集群搭 ...
- Kafka官方文档翻译——简介
简介 Kafka擅长于做什么? 它被用于两大类应用: 在应用间构建实时的数据流通道 构建传输或处理数据流的实时流式应用 几个概念: Kafka以集群模式运行在1或多台服务器上 Kafka以topics ...
- Kafka安装与简介
今天来讲一下Kafka,它是一个消息队列,应用场景比较广泛.刚开始学习一门东西,咱们先不管它是干什么的,先跑起来才是正经,所以本文主要讲两点: 安装搭建Kafka 简单介绍下Kafka的原理和应用 1 ...
- kafka入门:简介、使用场景、设计原理、主要配置及集群搭建--转载
问题导读: 1.zookeeper在kafka的作用是什么? 2.kafka中几乎不允许对消息进行"随机读写"的原因是什么? 3.kafka集群consumer和producer状 ...
最新文章
- php验证年月日的正则表达式,关于正则表达式:在php中验证日期格式
- Kali Linux NetHunter教程Kali NetHunter支持的设备和ROMs
- 华为交换机堆叠SVF助手(推荐)
- python3.6安装pip-python3.6如何安装pip
- Android开发工具之Android Studio--调用系统隐藏方法之操作aidl文件步骤
- Python接口测试-模块引用与映射
- GDCM:2个DICOM文件的diff测试程序
- clang 反汇编
- 林斌宣布好消息!语音识别大佬、Kaldi之父加盟小米
- 【分布式】Zookeeper序列化及通信协议
- Google推荐的图片加载库Glide介绍
- 《诗经》那么美,读不懂多浪费|爬取一本好好学习,准备做一个“有文化”的程序人
- 用ctrl+鼠标滚动调节字体大小
- ip 地址在线解析, api
- 【推荐算法】Knowledge-aware Graph Neural Networks with Label Smoothness Regularization(KGNN-LS)论文笔记
- puzzle(1024)帐篷、星之战
- springmvc案例
- Linux使用Geany开发gtk程序教程
- 基于SPI协议的Flash驱动控制-扇区擦除
- 最通俗易懂的python与mysql数据库交互
热门文章
- cordova报错:Could not find an installed version of Gradle either in Android Studio, or on your system
- Ansible搭建hadoop3.1.3高可用集群
- 编写Dockerfile增强centos镜像:安装vim 和net-tools
- 按照 排序 oracle,oracle 按照中文排序
- mysql sql执行加载顺序
- linux用户和权限管理
- redis持久化(2)
- Qt for Android获取手机序列号/手机型号/手机制造商
- Qt pro工程文件中判断宏定义是否存在
- 网络OSI和TCP/IP参考模型