引言

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

Kafka原理

Kafka 的设计初衷是希望做为一个统一的信息收集平台，能够实时的收集反馈信息，并需要能够支撑较大的数据量,且具备良好的容错能力。

Kafka 使用文件存储消息(append only log)，这就直接决定了 kafka 在性能上严重依赖文件系统的本身特性。为了减少磁盘写入的次数，broker 会将消息暂时缓存起来，当消息的个数(或尺寸)达到一定阀值时，再一起刷新到磁盘，这样会减少磁盘 IO 调用的次数。

Producer 将会和 topic 下所有 partition leader 保持 socket 连接。消息由 producer 直接通过 socket 发送到 broker，中间不会经过任何“路由层”。

事实上，消息被路由到哪个 partition 上，由 producer 客户端决定，默认方式为“轮询”。

Consumer 端向 broker 发送 “fetch” 请求，并告知其获取消息的 offset，此后 consumer 将会获得一定条数的消息。Consumer 端也可以重置 offset 来重新消费消息。

Kafka 将每个 partition 数据复制到多个 server 上，任何一个 partition 都有一个 leader 和多个 follower (可以没有)。

备份的个数可以通过 broker 配置文件来设定，其中 leader 处理所有的读写请求，follower 需要和 leader 保持同步。

当 leader 失效时，需在 followers 中重新选取出新的 leader，可能此时 follower 落后于 leader，因此需要选择一个 “up-to-date” 的 follower。选择 follower 时需要兼顾一个问题，就是新的 leader server 上所已经承载的 partition leader 的个数，如果一个 server 上有过多的 partition leader，意味着此 server 将承受着更多的 IO 压力，因此在选举新 leader 时，需要考虑到“负载均衡”。

Kafka 中所有的 topic 内容都是以日志的形式存储在 broker 上。如果一个 topic 的名称为 “my_topic”，它有 2 个 partitions，那么日志将会保存在 my_topic_0 和 my_topic_1 两个目录中。

日志文件中保存了一序列 “log entries” (日志条目)，每个 log entry 格式为“4个字节的数字 N 表示消息的长度” + “N 个字节的消息内容”。每个日志都有一个 offset 来唯一的标记一条消息，offset 的值为8个字节的数字，表示此消息在此 partition 中所处的起始位置。

kafka-manager安装部署

1、yum安装sbt(因为kafka-manager需要sbt编译)

[root@slave1 kafka]#curl https://bintray.com/sbt/rpm/rpm bintray-sbt-rpm.repo

[root@slave1 kafka]#mv bintray-sbt-rpm.repo /etc/yum.repos.d/

[root@slave1 kafka]#yum install sbt

2、下载kafka-manager并编译

[root@slave1 kafka]#git clone https://github.com/yahoo/kafka-managercd kafka-manager

[root@slave1 kafka]#cd kafka-manager

[root@slave1 kafka]#sbt clean distcd target/

[root@slave1 kafkamanager]# ll

total 58356

-rw-r--r-- 1 root root 59748895 Apr 14 11:32 kafka-manager-1.3.2.1.zip

3、解压缩kafka-manager

[root@slave1 kafkamanager]# unzip kafka-manager-1.3.2.1.zip

Archive: kafka-manager-1.3.2.1.zip

inflating: kafka-manager-1.3.2.1/lib/kafka-manager.kafka-manager-1.3.2.1-sans-externalized.jar

inflating: kafka-manager-1.3.2.1/lib/org.scala-lang.scala-library-2.11.8.jar

inflating: kafka-manager-1.3.2.1/lib/com.typesafe.play.twirl-api_2.11-1.1.1.jar

4、修改配置文件conf/application.properties

[root@slave1 kafka-manager-1.3.2.1]# cd conf/

[root@slave1 conf]# ll

total 24

-rw-r--r-- 1 root root 1291 Oct 13 2016 application.conf

-rw-r--r-- 1 root root 27 Oct 13 2016 consumer.properties

-rw-r--r-- 1 root root 2108 Oct 13 2016 logback.xml

-rw-r--r-- 1 root root 1367 Oct 13 2016 logger.xml

-rw-r--r-- 1 root root 7167 Oct 13 2016 routes

[root@slave1 conf]# vi application.conf

#zk地址

kafka-manager.zkhosts="slave1:2181,salve2:2181,salve3:2181"

#默认9000，由于我用的CDH，cdh占用了9000端口，所以更改了下

http.port=8080

5、启动

[root@slave1 kafka-manager-1.3.2.1]# bin/kafka-manager -Dconfig.file=conf/application.conf &

……

[info] k.m.a.DeleteClusterActor - Scheduling updater for 10 seconds

[info] k.m.a.KafkaManagerActor - Starting kafka manager path cache...

[info] k.m.a.KafkaManagerActor - Adding kafka manager path cache listener...

[info] play.api.Play - Application started (Prod)

[info] p.c.s.NettyServer - Listening for HTTP on /0:0:0:0:0:0:0:0:8080

……

6、登录webui查看

http://10.105.10.123:8080/

Paste_Image.png

出现这个界面表示安装成功。

总结

以 上就是我对Java大型互联网-Kafka原理设计实现之kafka-manager安装部署应用 问题及其优化总结，分享给大家，觉得收获的话可以点个关注收藏转发一波喔，谢谢大佬们支持！

最后，每一位读到这里的网友，感谢你们能耐心地看完。希望在成为一名更优秀的Java程序员的道路上，我们可以一起学习、一起进步！都能赢取白富美，走向架构师的人生巅峰!

想了解学习Java方面的技术内容以及Java技术视频的内容可加群：722040762 验证码：头条（06 必过）欢迎大家的加入哟！