一、熟悉kafka
l Server-1 broker其实就是kafka的server，因为producer和consumer都要去连它。Broker主要还是做存储用。
l Server-2是zookeeper的server端，zookeeper的具体作用你可以去官网查，在这里你可以先想象，它维持了一张表，记录了各个节点的IP、端口等信息（以后还会讲到，它里面还存了kafka的相关信息）。
l Server-3、4、5他们的共同之处就是都配置了zkClient，更明确的说，就是运行前必须配置zookeeper的地址，道理也很简单，这之间的连接都是需要zookeeper来进行分发的。
l Server-1和Server-2的关系，他们可以放在一台机器上，也可以分开放，zookeeper也可以配集群。目的是防止某一台挂了。
简单说下整个系统运行的顺序：
1. 启动zookeeper的server
2. 启动kafka的server
3. Producer如果生产了数据，会先通过zookeeper找到broker，然后将数据存放进broker
4. Consumer如果要消费数据，会先通过zookeeper找对应的broker，然后消费。
Kafka 分布式消息队列 类似产品有JBoss、MQ
一、由Linkedln 开源，使用scala开发，有如下几个特点：
（1）高吞吐
（2）分布式
（3）支持多语言客户端（C++、Java）
二、组成:客户端是 producer 和 consumer，提供一些API，服务器端是Broker，客户端提供可以向Broker内发布消息、消费消息，服务器端提供消息的存储等功能
Kafka 特点是支持分区、分布式、可拓展性强
三、Kafka的消息分几个层次
(1)Topic 一类主题
(2)Partition 默认每个消息有2个分区，创建Topic可以指定分区数，1天有 1亿行可以分8个分区，如果每天几十万行就一个分区吧
(3)Message 是每个消息
四、数据处理流程
1.生产者 生产消息、将消息发布到指定的topic分区
2.kafka 集群接收到producer发过来的消息后，将其持久化到硬盘，可以指定时长，而不关注消息是否被消费
3.consumer从kafka集群pull或push方式，并控制获取消息的offset偏移量，consumer重启时需要根据offset开始再次消费数据，consumer自己维护offset
五、kafka如何实现高吞吐量
1.充分利用磁盘的顺序读写
2.数据批量发送
3.数据压缩
4.Topic划分多个partition
六、kafka如何实现load balance &HA
1）producer 根据用户指定的算法，将消息发送到指定的partition
2）存在多个partition，每个partition存在多个副本replica，每个replica分布在不同的broker节点上
3）每个partition需要选取lead partition，leader partition负责读写，并由zookeeper负责fail over 快速失败
4）通过zookeeper管理broker与consumer的动态加入与离开
七、扩容
当需要增加broker节点时，新增的broker会向zookeeper注册，而producer及consumer会根据zookeeper上的watcher感知这些变化，并及时作出调整
副本分配逻辑规则如下：
在Kafka集群中，每个Broker都有均等分配Partition的Leader机会。
上述图Broker Partition中，箭头指向为副本，以Partition-0为例:broker1中parition-0为Leader，Broker2中Partition-0为副本。
上述图种每个Broker(按照BrokerId有序)依次分配主Partition,下一个Broker为副本，如此循环迭代分配，多副本都遵循此规则。
副本分配算法如下：
将所有N Broker和待分配的i个Partition排序.
将第i个Partition分配到第(i mod n)个Broker上.
将第i个Partition的第j个副本分配到第((i + j) mod n)个Broker上.
二、安装zookeeper，并配置集群
准备三台机器做集群
服务器
IP地址
端口
服务器1
172.16.0.41
2181/2881/3881
服务器2
172.16.0.42
2182/2882/3882
服务器3
172.16.0.43
2183/2883/3883
2.1配置java环境
将jdk-7u79-linux-x64上传到三台服务器安装配置。
给三台服务器分别创建java文件夹。
将jdk 放到java文件夹下并解压，然后删掉压缩文件。
配置jdk全局变量。
#vi /etc/profile
export JAVA_HOME=/usr/local/java/jdk1.7.0_79
export JRE_HOME=/usr/local/java/jdk1.7.0_79/jre
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib:$CLASSPATH
export PATH=$JAVA_HOME/bin:$PATH
2.2 修改操作系统的/etc/hosts文件，添加IP与主机名映射：
# zookeeper cluster servers
172.16.0.41 edu-zk-01
172.16.0.42 edu-zk-02
172.16.0.43 edu-zk-03
2.3下载zookeeper-3.4.7.tar.gz 到/home/zy/zookeeper目录
# mkdir -p /usr/local/zookeeper
# cd / usr/local/zookeeper/
# wgethttp://apache.fayea.com/zookeeper/zookeeper-3.4.7/zookeeper-3.4.7.tar.gz
2.4 解压zookeeper安装包，并对节点重民名
#tar -zxvf zookeeper-3.4.7.tar.gz
服务器1：
#mv zookeeper-3.4.7 node-01
服务器2：
#mv zookeeper-3.4.7 node-02
服务器3：
#mv zookeeper-3.4.7 node-03
2.5 在zookeeper的各个节点下 创建数据和日志目录
#cd /usr/local/zookeeper
#mkdir data
#mkdir logs
2.6 重命名配置文件
将zookeeper/node-0X/conf目录下的zoo_sample.cfg文件拷贝一份，命名为zoo.cfg:
#cp zoo_sample.cfg zoo.cfg
2.7 修改zoo.cfg 配置文件
三台服务器做同样配置：zookeeper/node-01的配置（/usr/local/zookeeper/node-01/conf/zoo.cfg）如下：
参数说明:
tickTime=2000
tickTime这个时间是作为Zookeeper服务器之间或客户端与服务器之间维持心跳的时间间隔,也就是每个tickTime时间就会发送一个心跳。
initLimit=10
initLimit这个配置项是用来配置Zookeeper接受客户端（这里所说的客户端不是用户连接Zookeeper服务器的客户端,而是Zookeeper服务器集群中连接到Leader的Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过10个心跳的时间（也就是tickTime）长度后Zookeeper 服务器还没有收到客户端的返回信息,那么表明这个客户端连接失败。总的时间长度就是10*2000=20 秒。
syncLimit=5
syncLimit这个配置项标识Leader与Follower之间发送消息,请求和应答时间长度,最长不能超过多少个tickTime的时间长度,总的时间长度就是5*2000=10秒。
dataDir=/usr/local/zookeeper/node-01/data
dataDir顾名思义就是Zookeeper保存数据的目录,默认情况下Zookeeper将写数据的日志文件也保存在这个目录里。
clientPort=2181
clientPort这个端口就是客户端（应用程序）连接Zookeeper服务器的端口,Zookeeper会监听这个端口接受客户端的访问请求。
server.A=B：C：D
server.1=edu-zk-01:2881:3881
server.2=edu-zk-02:2882:3882
server.3=edu-zk-03:2883:3883
A是一个数字,表示这个是第几号服务器；
B是这个服务器的IP地址（或者是与IP地址做了映射的主机名）；
C第一个端口用来集群成员的信息交换,表示这个服务器与集群中的Leader服务器交换信息的端口；
D是在leader挂掉时专门用来进行选举leader所用的端口。
注意：如果是伪集群的配置方式，不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号。
2.8 创建myid文件
在dataDir=/usr/local/zookeeper/node-0X/data 下创建myid文件
编辑myid文件，并在对应的IP的机器上输入对应的编号。如在node-01上，myid文件内容就是1,node-02上就是2，node-03上就是3：
#vi /usr/local/zookeeper/node-01/data/myid## 值为1
#vi /usr/local/zookeeper/node-02/data/myid## 值为2
#vi /usr/local/zookeeper/node-03/data/myid## 值为3
2.9 启动测试zookeeper
(1)进入/usr/local/zookeeper/node-0X/bin目录下执行：
#/usr/local/zookeeper/node-01/bin/zkServer.sh start
#/usr/local/zookeeper/node-02/bin/zkServer.sh start
#/usr/local/zookeeper/node-03/bin/zkServer.sh start
(2)输入jps命令查看进程：
其中，QuorumPeerMain是zookeeper进程，说明启动正常
(3)查看状态：
# /usr/local/zookeeper/node-01/bin/zkServer.sh status
(4)查看zookeeper服务输出信息：
由于服务信息输出文件在/usr/local/zookeeper/node-0X/bin/zookeeper.out
$ tail -500f zookeeper.out
三、KAFKA集群配置
利用安装zookeeper的三台服务器做KAFKA集群，也可以新建三个虚拟机去操作。
服务器
IP地址
端口
服务器1
172.16.0.41
9092
服务器2
172.16.0.42
9092
服务器3
172.16.0.43
9092
4.1 下载 kafka_2.9.2-0.8.1
分别在三台服务器创建kafka目录并且下载kafka压缩包
#mkdir /usr/local/kafka
#tar –zxvf kafka_2.9.2-0.8.1.tar.gz
4.2 创建log文件夹
#mkdir /usr/local/kafka/kafkalogs
4.3 配置kafka
#cd /usr/local/kafka/kafka_2.9.2-0.8.1/config
#vi server.properties 修改项如下：
broker.id=0 //当前机器在集群中的唯一标识
port=9092 //kafka对外提供服务的tcp端口
host.name=172.16.0.41 //主机IP地址
log.dirs=/usr/local/kafka/kafkalogs //log存放目录
message.max.byte=5048576 //kafka一条消息容纳的消息最大为多少
default.replication.factor=2 //每个分区默认副本数量
replica.fetch.max.bytes=5048576
zookeeper.connect=172.16.0.41:2181,172.16.0.42:2182,172.16.0.43:2183
4.4 启动kafka
# ./kafka-server-start.sh -daemon ../config/server.properties //后台启动运行
4.5 问题解决
[root@master ~]# /export/kafka/bin/kafka-console-producer.sh --broker-list 10.14.2.201:9092,10.14.2.202:9092,10.14.2.203:9092,10.14.2.204:9092 --topic test
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
# /export/kafka/bin/kafka-console-consumer.sh --zookeeper 10.14.2.201:2181,10.14.2.202:2181,10.14.2.203:2181,10.14.2.204:2181 --topic test --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
解决方法：
下载slf4j-1.7.6.zip
http://www.slf4j.org/dist/slf4j-1.7.6.zip
解压
unzip slf4j-1.7.6.zip
把slf4j-nop-1.7.6.jar 包复制到kafka libs目录下面
cd slf4j-1.7.6
cp slf4j-nop-1.7.6.jar /export/kafka/libs/
四、KAFKA集群验证
5.1 创建topic
#./kafka-topics.sh --create --zookeeper 172.16.0.42:2182 --replication-factor 1 --partitions 1 --topic test
5.2 查看topic
# ./kafka-topics.sh --list --zookeeper 172.16.0.42:2182
5.3 开启发送者并发送消息
#./kafka-console-producer.sh --broker-list 172.16.0.41:9092 --topic test
5.4 开启消费者并接收消息
#./kafka-console-consumer.sh --zookeeper 172.16.0.42:2182 --topic test --from-beginning
网友评论