文章目录
- Kafka介绍
- 1.Kafka的使用场景
- 2.Kafka基本概念
- kafka基本使用
- 1.安装前的环境准备
- 2.启动kafka服务器
- 3.创建主题topic
- 4.发送消息
- 5.消费消息
Kafka介绍
Kafka是最初由Linkedin公司开发,是一个分布式、支持分区的(partition)、多副本的
(replica),基于zookeeper协调的分布式消息系统,它的最大的特性就是可以实时的处理
大量数据以满足各种需求场景:比如基于hadoop的批处理系统、低延迟的实时系统、
Storm/Spark流式处理引擎,web/nginx日志、访问日志,消息服务等等,用scala语言编 写,Linkedin于 2010
年贡献给了Apache基金会并成为顶级开源 项目。
1.Kafka的使用场景
日志收集:一个公司可以用Kafka收集各种服务的log,通过kafka以统一接口服务的方式
开放给各种consumer,例如hadoop、Hbase、Solr等。 消息系统:解耦和生产者和消费者、缓存消息等。
用户活动跟踪:Kafka经常被用来记录web用户或者app用户的各种活动,如浏览网⻚、
搜索、点击等活动,这些活动信息被各个服务器发布到kafka的topic中,然后订阅者通过
订阅这些topic来做实时的监控分析,或者装载到hadoop、数据仓库中做离线分析和挖 掘。
运营指标:Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据,生产 各种操作的集中反馈,比如报警和报告。
2.Kafka基本概念
kafka是一个分布式的,分区的消息(官方称之为commit log)服务。它提供一个消息系统应该
具备的功能,但是确有着独特的设计。可以这样来说,Kafka借鉴了JMS规范的思想,但是确 并 没有完全遵循JMS规范。
首先,让我们来看一下基础的消息(Message)相关术语:
名称 | 解释 |
---|---|
Broker | 消息中间件处理节点,⼀个Kafka节点就是⼀个broker,⼀个或者多个Broker可以组成⼀个Kafka集群 |
Topic | Kafka根据topic对消息进⾏归类,发布到Kafka集群的每条消息都需要指定⼀个topic |
Producer | 消息⽣产者,向Broker发送消息的客户端 |
Consumer | 消息消费者,从Broker读取消息的客户端 |
ConsumerGroup | 每个Consumer属于⼀个特定的Consumer Group,⼀条消息可以被多个不同的Consumer Group消费,但是⼀个Consumer Group中只能有⼀个Consumer能够消费该消息 |
Partition | 物理上的概念,⼀个topic可以分为多个partition,每个partition内部消息是有序的 |
因此,从一个较高的层面上来看,producer通过网络发送消息到Kafka集群,然后consumer 来进行消费,如下图: 输入图片说明
服务端(brokers)和客户端(producer、consumer)之间通信通过 TCP协议 来完成。
kafka基本使用
1.安装前的环境准备
-
安装jdk
-
安装zk
-
官网下载kafka的压缩包:http://kafka.apache.org/downloads
解压缩至如下路径
/usr/local/kafka/
-
修改配置文件:/usr/local/kafka/kafka2.11-2.4/config/server.properties
#broker.id属性在kafka集群中必须要是唯一 broker.id= 0 #kafka部署的机器ip和提供服务的端口号 listeners=PLAINTEXT://192.168.65.60:9092 #kafka的消息存储文件 log.dir=/usr/local/data/kafka-logs #kafka连接zookeeper的地址 zookeeper.connect= 192.168.65.60:2181
2.启动kafka服务器
进入到bin目录下。使用命令来启动
./kafka-server-start.sh -daemon../config/server.properties
验证是否启动成功:
进入到zk中的节点看id是 0 的broker有没有存在(上线)
ls /brokers/ids/
server.properties核心配置详解:
Property | Default | Description |
---|---|---|
broker.id | 0 | 每个broker都可以⽤⼀个唯⼀的⾮负整数id进⾏标识;这个id可以作为broker的“名字”,你可以选择任意你喜欢的数字作为id,只要id是唯⼀的即可。 |
log.dirs | /tmp/kafka-logs | kafka存放数据的路径。这个路径并不是唯⼀的,可以是多个,路径之间只需要使⽤逗号分隔即可;每当创建新partition时,都会选择在包含最少partitions的路径下进⾏。 |
listeners | PLAINTEXT://192.168.65.60:9092 | server接受客户端连接的端⼝,ip配置kafka本机ip即可 |
zookeeper.connect | localhost:2181 | zooKeeper连接字符串的格式为:hostname:port,此处hostname和port分别是ZooKeeper集群中某个节点的host和port;zookeeper如果是集群,连接⽅式为hostname1:port1, hostname2:port2,hostname3:port3 |
log.retention.hours | 168 | 每个⽇志⽂件删除之前保存的时间。默认数据保存时间对所有topic都⼀样。 |
num.partitions | 1 | 创建topic的默认分区数 |
default.replication.factor | 1 | ⾃动创建topic的默认副本数量,建议设置为⼤于等于2 |
min.insync.replicas | 1 | 当producer设置acks为-1时,min.insync.replicas指定replicas的最⼩数⽬(必须确认每⼀个repica的写数据都是成功的),如果这个数⽬没有达到,producer发送消息会产⽣异常 |
delete.topic.enable | false | 是否允许删除主题 |
3.创建主题topic
topic是什么概念?topic可以实现消息的分类,不同消费者订阅不同的topic。
执行以下命令创建名为“test”的topic,这个topic只有一个partition,并且备份因子也设置为1
./kafka-topics.sh --create --zookeeper 172.16.253.35:2181 --replication-factor 1 --partitions 1 --topic test
查看当前kafka内有哪些topic
./kafka-topics.sh --list --zookeeper 172.16.253.35:2181
4.发送消息
kafka自带了一个producer命令客户端,可以从本地文件中读取内容,或者我们也可以以命令行中直接输入内容,并将这些内容以消息的形式发送到kafka集群中。在默认情况下,每一个行会被当做成一个独立的消息。使用kafka的发送消息的客户端,指定发送到的kafka服务器地址和topic
./kafka-console-producer.sh --broker-list 172.16.253.38:9092 --topic test
5.消费消息
对于consumer,kafka同样也携带了一个命令行客户端,会将获取到内容在命令中进行输 出, 默认是消费最新的消息
。使用kafka的消费者消息的客户端,从指定kafka服务器的指定 topic中消费消息
方式一:从最后一条消息的偏移量+1开始消费
./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092 --topic test
方式二:从头开始消费
./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092 --from-beginning --topic test
几个注意点:
- 消息会被存储
- 消息是顺序存储
- 消息是有偏移量的
- 消费时可以指明偏移量进行消费