Kafka原理及单机部署

博文大纲：
1、kafka介绍及原理
2、部署单机kafka

1、kafka介绍及原理

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

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

一、kafka的特性

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

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

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

三、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两天。

四、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级别的复制的。

五、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填充。

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

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

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

七、leader选举

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

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

2、部署单机kafka

一、部署kafka

kafka服务依赖JAVA环境，我这里默认有。

kafka的安装包能够从个人网盘连接中下载。

#解包
[root@kafka src]# tar zxf kafka_2.11-2.2.1.tgz 
[root@kafka src]# mv kafka_2.11-2.2.1 /usr/local/kafka
[root@kafka src]# 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 -anpt | grep 9092  #肯定端口在监听

因为kafka是经过zookeeper来调度的，因此，即便是单机kafka也须要启动zookeeper服务，kafka的安装目录下是默认集成了zookeeper的，直接启动便可。

二、测试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
[root@kafka bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic test
>aaaa
>bbbb
>cccc
#开启新的终端，进行读取消息测试，“--from-beginning”表示从开头读取
[root@kafka bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
aaaa
bbbb
cccc

———————— 本文至此结束，感谢阅读 ————————