一、kafka介绍及原理

kafka是由Apache软件基金会发布的一个开源流处理平台，由Scala和Java编写。它是一种高吞吐量的分布式发布的订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。

这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

> 1、kafka的特性

kafka是一种高吞吐量的分布式发布订阅消息系统，具有以下特性：

通过磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能；
持久性：使用文件性存储，日志文件存储消息，需要写入硬盘，采用达到一定阈值才写入硬盘，从而减少磁盘I/O，如果kafka突然宕机，数据会丢失一部分；
高吞吐量：即使是非常普通的硬件kafka也可以支持每秒数百万的消息；
支持通过kafka服务器和消费机集群来分区消息；
支持Hadoop并行数据加载。

> 2、kafka相关术语

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内部是有序的。

> 3、Topic和Partition的区别

一个topic可以认为一个一类消息，每个topic将被分成多个partition，每个partition在存储层面是append log文件。任何发布到此partition的消息都会被追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为long型的数字，它唯一标记一条消息。每条消息都被append到partition中，是顺序写磁盘，因此效率非常高（顺序写磁盘比随机写内存的速度还要高，这是kafka高吞吐率的一个很重要的保证）。

每一条消息被发送到broker中，会根据partition规则选择被存储到哪一个partition（默认采用轮询的方式进行写入数据）。如果partition规则设置合理，所有消息可以均匀分布到不同的partition里，这样就实现了水平扩展。（如果一个topic对应一个文件，那这个文件所在的机器I/O将会成为这个topic的性能瓶颈，而partition解决了这个问题），如果消息被消费则保留append.log两天

> 4、kafka的架构

如上图所示，一个典型的kafka体系架构包括若干Producer（可以是服务器日志，业务数据，页面前端产生的page view等），若干个broker（kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer（Group），以及一个Zookeeper集群。kafka通过Zookeeper管理集群配置，选举出leader，以及在consumer group发生变化时进行重新调整。Producer使用push（推）模式将消息发布到broker，consumer使用pull（拉）模式从broker订阅并消费消息。

zookeeper群集中有两个角色：leader和follower，leader对外提供服务，follower负责leader里面所产生内容同步消息写入生成时产生replicas（副本）；
kafka的高可靠性的保证来源于其健壮的副本（replicas）策略。通过调节其副本相关参数，可以使得kafka在性能和可靠性之间运转之间的游刃有余。kafka从0.8.x版本开始提供partition级别的复制的。
5、kafka的文件存储机制
kafka中消息是以topic进行分类的，生产者通过topic向kafka broker发送消息，消费者通过topic读取数据。然而topic在物理层面又能以partition为分组，一个topic可以分为若干个partition，partition还可以细分为segment，一个partition物理上由多个segment组成。

为了便于说明问题，假设这里只有一个kafka集群，且这个集群只有一个kafka broker，也就是只有一台物理机。在这个kafka broker的server.properties配置文件中定义kafka的日志文件存放路径以此来设置kafka消息文件存储目录，与此同时创建一个topic：test，partition的数量为4，启动kafka就可以在日志存放路径中看到生成4个目录，在kafka文件存储中，同一个topic下有多个不同的partition，每个partition为一个目录，partition的名称规则为：topic名称+有序序号，第一个序号从0开始。

segment是什么？

如果就以partition为最小存储单位，我们可以想象当Kafka producer不断发送消息，必然会引起partition文件的无限扩张，这样对于消息文件的维护以及已经被消费的消息的清理带来严重的影响，所以这里以segment为单位又将partition细分。每个partition(目录)相当于一个巨型文件被平均分配到多个大小相等的segment(段)数据文件中（每个segment 文件中消息数量不一定相等）这种特性也方便old segment的删除，即方便已被消费的消息的清理，提高磁盘的利用率。每个partition只需要支持顺序读写就行。

segment文件由两部分组成，分别为“.index”文件和“.log”文件，分别表示为segment索引文件和数据文件。这两个文件的命令规则为：partition全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值（偏移量），数值大小为64位，20位数字字符长度，没有数字用0填充。

> 6、数据的可靠性和持久性保证

当producer向leader发送数据时，可以通request.required.acks参数来设置数据可靠性的级别：

1（默认）：producer的leader已成功收到数据并得到确认。如果leader宕机了，则会丢失数据；
0 ：producer无需等待来自broker的确认而继续发送下一批消息。这种情况下数据传输效率最高，但是数据可靠性确是最低的；
-1：producer需要等待所有follower都确认接收到数据后才算一次发送完成，可靠性最高。

> 7、leader选举

一条消息只有被所有follower都从leader复制过去才会被认为已提交。这样就避免了部分数据被写进了leader，还没来得及被任何follower复制就宕机了，而造成数据丢失。而对于producer而言，它可以选择是否等待消息commit。

一种非常常用的选举leader的方式是“少数服从多数”，在进行数据的复制过程中，存在多个follower，并且每个follower的数据速度都不相同，当leader宕机后，当前的follower上谁的数据最多谁就是leader。

> 二、部署单机kafka

1、部署kafka

kafka服务依赖于JAVA环境，如果没有，自行安装
[root@kafka ~]# wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.11-2.2.1.tgz
[root@kafka ~]# tar zxf kafka_2.11-2.2.1.tgz
[root@kafka ~]# mv kafka_2.11-2.2.1/ /usr/local/kafka
[root@kafka ~]# cd /usr/local/kafka/bin/

启动zookeeper

[root@kafka bin]# ./zookeeper-server-start.sh ../config/zookeeper.properties &

启动kafka

[root@kafka bin]# ./kafka-server-start.sh ../config/server.properties &

查看端口是否在监听

[root@kafka bin]# netstat -anput | grep 9092
tcp6 0 0 :::9092 :::* LISTEN 43326/java
tcp6 0 0 192.168.171.134:44388 192.168.171.134:9092 ESTABLISHED 43326/java
tcp6 0 0 192.168.171.134:9092 192.168.171.134:44388 ESTABLISHED 43326/java

由于kafka是通过zookeeper来调度的，所以，即使是单机kafka也需要启动zookeeper服务，kafka的安装目录下是默认集成了zookeeper的，直接启动即可。
2、测试kafka

在本机创建kafka，副本数量为1，分区数量为1

[root@kafka bin]# ./kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

查看本机的topic

[root@kafka bin]# ./kafka-topics.sh --list --bootstrap-server localhost:9092
test

发送消息到test

[root@kafka bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic test

aaa
bbb
ccc

开启新的终端，进行读取消息测试，“--from-beginning”表示从开头读取

[root@kafka bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
aaa
bbb
ccc