KafKa（一）安装、入门

1、KafKa是什么

在流式计算中&＃xff0c;Kafka一般用来缓存数据&＃xff0c;SparkStreaming、Flink通过消费Kafka的数据进行计算。

1. Apache Kafka是一个开源流平台&＃xff0c;使用Scala开发。是由Apache软件基金会开发的一个开源消息系统项目。

2. Kafka最初是由LinkedIn公司开发&＃xff0c;并于 2011年初开源。2012年10月从Apache Incubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。

3. Kafka是一个分布式消息队列。Kafka对消息保存时根据Topic进行归类&＃xff0c;发送消息者称为Producer&＃xff0c;消息接受者称为Consumer&＃xff0c;此外kafka集群有多个kafka实例组成&＃xff0c;每个实例(server)成为broker。

4. 无论是kafka集群&＃xff0c;还是producer和consumer都依赖于zookeeper集群保存一些meta信息&＃xff0c;来保证系统可用性。

2、消息队列内部实现原理

1.点对点模式

&＃xff08;一对一&＃xff0c;消费者主动拉取数据&＃xff0c;消息收到后消息清除&＃xff09;

点对点模型通常是一个基于拉取或者轮询的消息传送模型&＃xff0c;这种模型从队列中请求信息&＃xff0c;而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理&＃xff0c;即使有多个消息监听者也是如此。

2.发布/订阅模式

&＃xff08;一对多&＃xff0c;数据生产后&＃xff0c;推送给所有订阅者&＃xff09;

发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者&＃xff0c;临时订阅者只在主动监听主题时才接收消息&＃xff0c;而持久订阅者则监听主题的所有消息&＃xff0c;即使当前订阅者不可用&＃xff0c;处于离线状态。

3、为什么需要消息队列

1&＃xff09;解耦&＃xff1a;

　　允许你独立的扩展或修改两边的处理过程&＃xff0c;只要确保它们遵守同样的接口约束。

2&＃xff09;冗余&＃xff1a;

消息队列把数据进行持久化直到它们已经被完全处理&＃xff0c;通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中&＃xff0c;在把一个消息从队列中删除之前&＃xff0c;需要你的处理系统明确的指出该消息已经被处理完毕&＃xff0c;从而确保你的数据被安全的保存直到你使用完毕。

3&＃xff09;扩展性&＃xff1a;

因为消息队列解耦了你的处理过程&＃xff0c;所以增大消息入队和处理的频率是很容易的&＃xff0c;只要另外增加处理过程即可。

4&＃xff09;灵活性 & 峰值处理能力&＃xff1a;

在访问量剧增的情况下&＃xff0c;应用仍然需要继续发挥作用&＃xff0c;但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力&＃xff0c;而不会因为突发的超负荷的请求而完全崩溃。

5&＃xff09;可恢复性&＃xff1a;

系统的一部分组件失效时&＃xff0c;不会影响到整个系统。消息队列降低了进程间的耦合度&＃xff0c;所以即使一个处理消息的进程挂掉&＃xff0c;加入队列中的消息仍然可以在系统恢复后被处理。

6&＃xff09;顺序保证&＃xff1a;

在大多使用场景下&＃xff0c;数据处理的顺序都很重要。大部分消息队列本来就是排序的&＃xff0c;并且能保证数据会按照特定的顺序来处理。&＃xff08;Kafka保证一个Partition内的消息的有序性&＃xff09;

7&＃xff09;缓冲&＃xff1a;

有助于控制和优化数据流经过系统的速度&＃xff0c;解决生产消息和消费消息的处理速度不一致的情况。

8&＃xff09;异步通信&＃xff1a;

很多时候&＃xff0c;用户不想也不需要立即处理消息。消息队列提供了异步处理机制&＃xff0c;允许用户把一个消息放入队列&＃xff0c;但并不立即处理它。想向队列中放入多少消息就放多少&＃xff0c;然后在需要的时候再去处理它们。

4、Kafka架构

1&＃xff09;Producer &＃xff1a;消息生产者&＃xff0c;就是向kafka broker发消息的客户端。

2&＃xff09;Consumer &＃xff1a;消息消费者&＃xff0c;向kafka broker取消息的客户端

3&＃xff09;Topic &＃xff1a;可以理解为一个队列。

4&＃xff09; Consumer Group &＃xff08;CG&＃xff09;&＃xff1a;这是kafka用来实现一个topic消息的广播&＃xff08;发给所有的consumer&＃xff09;和单播&＃xff08;发给任意一个consumer&＃xff09;的手段。一个topic可以有多个CG。topic的消息会复制-给consumer。如果需要实现广播&＃xff0c;只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。

5&＃xff09;Broker &＃xff1a;一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。

6&＃xff09;Partition&＃xff1a;为了实现扩展性&＃xff0c;一个非常大的topic可以分布到多个broker&＃xff08;即服务器&＃xff09;上&＃xff0c;一个topic可以分为多个partition&＃xff0c;每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id&＃xff08;offset&＃xff09;。kafka只保证按一个partition中的顺序将消息发给consumer&＃xff0c;不保证一个topic的整体&＃xff08;多个partition间&＃xff09;的顺序。

7&＃xff09;Offset&＃xff1a;kafka的存储文件都是按照offset.kafka来命名&＃xff0c;用offset做名字的好处是方便查找。例如你想找位于2049的位置&＃xff0c;只要找到2048.kafka的文件即可。当然the topic1 offset就是00000000000.kafka

1、集群规划

zhaohui01 zhaohui02 zhaohui03

zookeeper zookeeper zookeeper

kafka kafka kafka

2、安装Zookeeper

1、下载安装包zookeeper-3.5.7

2、上传至集群 /opt/ 解压安装包

&＃xff08;1&＃xff09;在 /opt/zookeeper/ 这个目录下创建 data 文件夹

&＃xff08;2&＃xff09;重命名 /opt/zookeeper/conf 这个目录下的 zoo_sample.cfg 为 zoo.cfg

tar -txvf apache-zookeeper-3.5.7-bin.tar.gz zookeeper-3.5.7

3、修改配置文件

(1)配置zoo.cfg

#增加如下配置
dataDir &＃61; /opt/zookeeper-3.5.7/data
​
server.1 &＃61; zhaohui01:2888:3888
server.2 &＃61; zhaohui02:2888:3888
server.3 &＃61; zhaohui03:2888:3888

配置 zkEnv.sh 文件

export JAVA_HOME&＃61;/opt/jdk-1.8

配置 /etc/profile 文件

export ZOO_HOME&＃61;/opt/zookeeper-3.5.7
export PATH&＃61;$ZOO_HOME/bin:$PATH

&＃xff08;2&＃xff09;配置参数解读

Server.A&＃61;B:C:D。

A是一个数字&＃xff0c;表示这个是第几号服务器&＃xff1b;

B是这个服务器的ip地址&＃xff1b;

C是这个服务器与集群中的Leader服务器交换信息的端口&＃xff1b;

D是万一集群中的Leader服务器挂了&＃xff0c;需要一个端口来重新进行选举&＃xff0c;选出一个新的Leader&＃xff0c;而这个端口就是用来执行选举时服务器相互通信的端口。

集群模式下配置一个文件myid&＃xff0c;这个文件在dataDir目录下&＃xff0c;这个文件里面有一个数据就是A的值&＃xff0c;Zookeeper启动时读取此文件&＃xff0c;拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。

(3)集群操作

1.在 /opt/zookeeper-3.5.7/data 目录下创建一个 myid 的文件

touch myid

添加myid文件&＃xff0c;注意一定要在linux里面创建&＃xff0c;在notepad&＃43;&＃43;里面很可能乱码

2.编辑 myid 文件

vim myid

在文件中添加与server对应的编号&＃xff1a;如1

3.拷贝配置好的zookeeper到其他机器上

分别修改myid文件中内容为2、3

&＃xff08;4&＃xff09;分发zookeeper文件

(5)编写zookeeper脚本到/root/shell/

zk.sh

#!/bin/bash
​
case $1 in
"start"){for i in zhaohui01 zhaohui02 zhaohui03doecho --------- zookeeper $i 启动-----------ssh $i "/opt/zookeeper-3.5.7/bin/zkServer.sh start"doneecho --------- zookeeper启动完成-----------
}
;;
"stop"){
for i in zhaohui01 zhaohui02 zhaohui03doecho --------- zookeeper $i 停止-----------ssh $i "/opt/zookeeper-3.5.7/bin/zkServer.sh stop"doneecho --------- zookeeper停止完成-----------
}
;;
"status"){
for i in zhaohui01 zhaohui02 zhaohui03doecho --------- zookeeper $i 状态-----------ssh $i "/opt/zookeeper-3.5.7/bin/zkServer.sh status"done
​
}
;;
esac
​

启动测试&＃xff01;

注意&＃xff1a;

zookeeper启动数量要 >&＃61;2 ,而且只有一个leader&＃xff0c;其他都是follower&＃xff0c; 而且在leader存活时不会找其他的follower&＃xff0c;当leader死了后&＃xff0c;follower代替leader工作&＃xff0c;成为第二个leader

3、安装 KafKa

1、下载安装包 kafka_2.11-2.4.1.tgz

2、上传解压安装包

&＃xff08;1&＃xff09;在 /opt/kafka-2.4.1/ 这个目录下创建 data 文件夹

tar -zxvf kafka_2.11-2.4.1.tgz

3、修改配置文件

vim /opt/kafka-2.4.1/confing/server.properties

输入以下内容&＃xff1a;

#broker的全局唯一编号&＃xff0c;不能重复
broker.id&＃61;0
#是否允许删除topic
delete.topic.enable&＃61;true
#kafka数据日志存放的路径
log.dirs&＃61;/opt/kafka-2.4.1/data
#segment文件保留的最长时间&＃xff0c;超时将被删除
log.retention.hours&＃61;168
#配置连接Zookeeper集群地址
zookeeper.connect&＃61;zhaohui01:2181,zhaohui02:2181,zhaohui03:2181

配置环境变量

export KAFKA_HOME&＃61;/opt/kafka-2.4.1
export PATH&＃61;$KAFKA_HOME/bin:$PATH

资源刷新

分发安装包到其他机器

分别在zhaohui02和zhaohui03上修改配置文件/opt/kafka-2.4.1/config/server.properties中的broker.id&＃61;1、broker.id&＃61;2

注&＃xff1a;broker.id不得重复

4、编写kafka脚本

#!/bin/bash
​
BROKERS&＃61;"zhaohui01 zhaohui02 zhaohui03"
# Kafka的安装目录
KAFKA_HOME&＃61;"/opt/kafka-2.4.1"
​
case $1 in
"start"){
for broker in $BROKERSdoecho "----------- 正在启动 kafka on ${broker} ----------"ssh $broker  "source /etc/profile;nohup ${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties >/dev/null2>&1 &"if [ $? !&＃61; 0 ];thenecho "Can not starting kafka server on host ${broker}";exit 1; fi
doneecho ----------- kafka 集群启动完成 -----------
}
;;
"stop"){
for broker in $BROKERSdoecho --------- 正在停止 kafka on ${broker} -----------ssh $broker "source /etc/profile;${KAFKA_HOME}/bin/kafka-server-stop.sh  ${KAFKA_HOME}/config/server.properties"doneecho ----------- kafka 集群停止完成 -----------
}
;;
esac

4、KafKa 命令行操作

1.查看当前服务器中的所有topic&＃xff08;主题&＃xff09;

[root&＃64;zhaohui01 ~]# kafka-topics.sh --list --zookeeper zhaohui01:2181

2.创建topic

[root&＃64;zhaohui01 ~]# kafka-topics.sh --create --zookeeper zhaohui01:2181 --topic first02 --partitions 3 --replication-factor 1

选项说明&＃xff1a;

--topic 定义topic名

--replication-factor 定义副本数

--partitions 定义分区数

注意&＃xff1a;副本数量不能超过机器数量

3.删除topic

[root&＃64;zhaohui01 ~]# kafka-topics.sh --zookeeper node01:2181 --delete --topic first0

需要server.properties中设置delete.topic.enable&＃61;true否则只是标记删除。