kafka

kafka概述

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源 项目。

当今社会各种应用系统诸如商业、社交、搜索、浏览等像信息工厂一样不断的生产出各种信息，在大数据时代，我们面临如下几个挑战：

（1）如何收集这些巨大的信息

（2）如何分析它

（3）如何及时做到如上两点

以上几个挑战形成了一个业务需求模型，即生产者生产（produce）各种信息，消费者消费（consume）（处理分析）这些信息，而在生产者与消费者之间，需要一个沟通两者的桥梁-消息系统。从一个微观层面来说，这种需求也可理解为不同的系统之间如何传递消息。

消息队列通信模式

1. 点对点模式

点对点模式通常是基于拉取或者轮询的消息传送模型，这个模型的特点是发送到队列的消息被一个且只有一个消费者进行处理。生产者将消息放入消息队列后，由消费者主动的去拉取消息进行消费。点对点模型的的优点是消费者拉取消息的频率可以由自己控制。但是消息队列是否有消息需要消费，在消费者端无法感知，所以在消费者端需要额外的线程去监控。

2.发布订阅模式

发布订阅模式是一个基于消息传送模型，该模型可以有多种不同的订阅者。生产者将消息放入消息队列后，队列会将消息推送给订阅过该类消息的消费者（类似微信公众号）。由于是消费者被动接收推送，所以无需感知消息队列是否有待消费的消息！但是consumer1、consumer2、consumer3由于机器性能不一样，所以处理消息的能力也会不一样，但消息队列却无法感知消费者消费的速度！所以推送的速度成了发布订阅模模式的一个问题！假设三个消费者处理速度分别是8M/s、5M/s、2M/s，如果队列推送的速度为5M/s，则consumer3无法承受！如果队列推送的速度为2M/s，则consumer1、consumer2会出现资源的极大浪费！

kafka的架构原理

Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据，具有高性能、持久化、多副本备份、横向扩展能力

Producer：Producer即生产者，消息的产生者，是消息的入口。

Broker：Broker是kafka实例，每个服务器上有一个或多个kafka的实例，我们姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号，如图中的broker-0、broker-1等……

Topic：消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。

Partition：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的文件夹！

Replication:每一个分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。

Message：每一条发送的消息主体。

Consumer：消费者，即消息的消费方，是消息的出口。

Consumer Group：我们可以将多个消费组组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！

Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。

消息写入leader后，follower是主动的去leader进行同步的！producer采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的！

kafka部署

//  环境准备
//  192.168.35.142    zookeeper1
//  192.168.35.143    zookeeper2
//  192.168.35.144    zookeeper3

//zookeeper1安装java
[root@zookeeper1 software]# ls
jdk-8u181-linux-x64.tar.gz
[root@zookeeper1 software]# tar -zxvf jdk-8u181-linux-x64.tar.gz
[root@zookeeper1 software]# vim /etc/profile
export JAVA_HOME=/opt/software/jdk1.8.0_181
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

//将java目录和/etc/profile都发送到zookeeper2和zookeeper3
[root@zookeeper1 software]# scp  -r  jdk1.8.0_181/ [email protected]:/opt/software/
[root@zookeeper1 software]# scp  -r  jdk1.8.0_181/ [email protected]:/opt/software/
[root@zookeeper1 software]# scp /etc/profile [email protected]:/etc/profile
[root@zookeeper1 software]# scp /etc/profile [email protected]:/etc/profile

//三台主机全部都source一下/etc/profile
[root@zookeeper1 ~]# source /etc/profile
[root@zookeeper2 ~]# source /etc/profile
[root@zookeeper3 ~]# source /etc/profile

//zookeeper1安装zookeeper
[root@zookeeper1 software]# tar -zxvf zookeeper-3.4.8.tar.gz
[root@zookeeper1 software]# mv zookeeper-3.4.8 zookeeper
[root@zookeeper1 software]# cd zookeeper/
[root@zookeeper1 zookeeper]# mkdir data logs
[root@zookeeper1 zookeeper]# cd conf/
[root@zookeeper1 conf]# cp zoo_sample.cfg zoo.cfg
[root@zookeeper1 conf]# vim zoo.cfg
dataDir=/opt/software/zookeeper/data //修改 dataDir 参数内容
//在文档最末尾填写如下几行
server.1=192.168.35.142:2888:3888
server.2=192.168.35.143:2888:3888
server.3=192.168.35.144:2888:3888

//在每个节点写入该节点的标识编号，每个节点编号不同，zookeeper1写入 1， zookeeper2写入 2，zookeeper3写入 3
[root@zookeeper1 conf]# echo 1 > /opt/software/zookeeper/data/myid
//将zookeeper发送给另外两台主机
[root@zookeeper1 software]# scp -r zookeeper/ [email protected]:/opt/software/
[root@zookeeper1 software]# scp -r zookeeper/ [email protected]:/opt/software/

//在zookeeper2中修改,在zookeeper3中修改
[root@zookeeper2 ~]# echo 2 > /opt/software/zookeeper/data/myid
[root@zookeeper3 ~]# echo 3 > /opt/software/zookeeper/data/myid

//配置zookeeper的环境变量
[root@zookeeper1 ~]# vim /etc/profile
export JAVA_HOME=/opt/software/jdk1.8.0_181
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export ZOOKEEPER_HOME=/opt/software/zookeeper
export PATH=$PATH:$ZOOKEEPER_HOME/bin

//将/etc/profile文件发送给另外两台主机
[root@zookeeper1 software]# scp /etc/profile [email protected]:/etc/profile
[root@zookeeper1 software]# scp /etc/profile [email protected]:/etc/profile

//三台主机全部都source一下/etc/profile
[root@zookeeper1 ~]# source /etc/profile
[root@zookeeper2 ~]# source /etc/profile
[root@zookeeper3 ~]# source /etc/profile

//三台主机都启动zookeeper
[root@zookeeper1 ~]# zkServer.sh start
[root@zookeeper2 ~]# zkServer.sh start
[root@zookeeper3 ~]# zkServer.sh start

//三台主机都查询一下zookeeper的状态
[root@zookeeper1 ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/software/zookeeper/bin/../conf/zoo.cfg
Mode: follower

[root@zookeeper2 ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/software/zookeeper/bin/../conf/zoo.cfg
Mode: follower

[root@zookeeper3 ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/software/zookeeper/bin/../conf/zoo.cfg
Mode: leader

//zookeeper1安装kafka
[root@zookeeper1 ~]# tar -zxvf kafka_2.11-2.4.0.tgz

//编辑kafka的配置文件
[root@zookeeper1 ~]# vim kafka_2.11-2.4.0/config/server.properties
在配置文件中找到以下两行并注释掉（在文本前加#）如下所示：
#broker.id=0
#zookeeper.connect=localhost:2181
添加
broker.id=1
zookeeper.connect=192.168.35.142:2181,192.168.35.143:2181,192.168.35.144:2181
listeners = PLAINTEXT://192.168.35.142:9092

//将kafka发送给另外两台主机
[root@zookeeper1 software]# scp kafka_2.11-2.4.0/ [email protected]:/etc/profile
[root@zookeeper1 software]# scp kafka_2.11-2.4.0/ [email protected]:/etc/profile

//修改zookeeper2主机的kafka配置文件
[root@zookeeper2 ~]# vim kafka_2.11-2.4.0/config/server.properties
broker.id=2
zookeeper.connect=192.168.35.142:2181,192.168.35.143:2181,192.168.35.144:2181
listeners = PLAINTEXT://192.168.35.143:9092

//修改zookeeper3主机的kafka配置文件
[root@zookeeper3 ~]# vim kafka_2.11-2.4.0/config/server.properties
broker.id=3
zookeeper.connect=192.168.35.142:2181,192.168.35.143:2181,192.168.35.144:2181
listeners = PLAINTEXT://192.168.35.144:9092

//三台主机全部启动kafka
[root@zookeeper1 ~]# ./kafka_2.11-2.4.0/bin/kafka-server-start.sh -daemon ./kafka_2.11-2.4.0/config/server.properties

[root@zookeeper2 ~]# ./kafka_2.11-2.4.0/bin/kafka-server-start.sh -daemon ./kafka_2.11-2.4.0/config/server.properties

[root@zookeeper3 ~]# ./kafka_2.11-2.4.0/bin/kafka-server-start.sh -daemon ./kafka_2.11-2.4.0/config/server.properties

//jps查看
[root@zookeeper1 ~]# jps
2770 Jps
2024 Kafka
1565 QuorumPeerMain

[root@zookeeper2 ~]# jps
1905 Kafka
2339 Jps
1452 QuorumPeerMain

[root@zookeeper3 ~]# jps
1155 QuorumPeerMain
1704 Kafka
2172 Jps

//测试kafka
[root@zookeeper1 ~]# ./kafka_2.11-2.4.0/bin/kafka-topics.sh --create --zookeeper 192.168.35.142:2181 --replication-factor 1 --partitions 1 --topic test

[root@zookeeper2 ~]# ./kafka_2.11-2.4.0/bin/kafka-topics.sh  --list --zookeeper 192.168.35.143:2181
test

[root@zookeeper3 ~]# ./kafka_2.11-2.4.0/bin/kafka-topics.sh  --list --zookeeper 192.168.35.144:2181
test