1. 初识Kafka

Kafka 最初是由微软旗下LinkedIn(领英)公司采用Scale 语言开发开发的
分布式消息系统来捐献给了Apache基金会。

Kafka 其实是一个分布式流处理平台，但是由于它拥有这样几个功能让它变得独特

• 1.消息流订阅和发布

• 流可以理解成字节流，TCP/IP Socket 通信传输数据包大多是字节流。
• 这个功能也就是说它拥有传统消息中间件的消息订阅和发布功能

• 2.消息流处理
• 3.分布式流存储

看到没？还拥有分布式存储的功能，所以它比传统的消息中间件要更加强大。

总结就是：

• Kafka 是一个分布式流处理框架
• 可以像一个消息中间件一样读写数据流（即，发布和订阅消息）
• Kafka 还拥有分布式流处理以及存储到磁盘的功能，所以比消息中间件更加强大。

首先检查下Kafka Linux 操作系统的信息

uname -a

执行结果如下所示：

当前我使用的Linux 操作系统版本是Centos7

关闭防火墙服务命令

systemctl stop firewalld.service

设置防火墙开放端口
有时候由于Linux 服务器上防火墙会端口拦截，所以我们需要在防火墙中开放80 端口，方便我们待会在物理机上查看

firewall-cmd --permanent --add-port=80/tcp --zone=public

重载防火墙配置

firewall-cmd --reload

由于Kafka 是基于JVM上运行的，因此我们需要安装JRE环境（JDK是JRE的超集，包含了JRE和一些开发工具包）。
管理员身份安装JDK 8

su -c "yum install java-1.8.0-openjdk-devel.x86_64"

安装完成后，我们使用jps命令会发现找不到，这是因为centos 7 安装的是精简版的open jdk,因此我们还需要安装一个JDK的插件

从yum 仓库中查找带插件的jdk

yum list | grep jdk-devel

执行结果如下：

查看open jdk 版本

rpm -qa | grep openjdk

执行结果如下：

查看JDK 安装路径

whereis java

执行结果如下所示：

查看JDK版本

java -version

执行结果如下所示：

貌似通过yum安装完JDK后环境变量不需要配置也可以直接使用

Zookeeper 是安装Kafka集群的必要组件，因为Kafka是通过Zookeeper 实现集群管理和leader选举的，因此我们还需要安装Zookeeper.

Zookeerp 官网：http://zookeeper.apache.org/ Zookeeper
下载地址：http://zookeeper.apache.org/releases.html#download

Zookeeper 有三个角色:

• leader: 主用者，领导者
• follower：备用者,关注者
• observer:观察者，旁观者

作用：一个集群最多只有一个Leader多个follower以及observer. 当leader服务器挂了的时候，可以从follower中选举出一个新的Leader.

下载完成后放到/opt 文件夹下执行解压命令

tar zxvf  zookeeper-3.4.12.tar.gz

解压之后会生成一个zookeeper-3.4.12文件夹

打开环境变量配置文件

vi /etc/profile

进入VI编辑器模式

i

添加环境变量

#zookeeper cofig var
export ZOOKEEPER_HOME=/opt/zookeeper-3.4.12
export PATH=$PATH:$ZOOKEEPER_HOME/bin

修改完成后，退出编辑模式按键盘上的ESC
保存并退出输入命令

:wq

使用source让环境变量即时生效

source /etc/profile

进入解压的文件夹

进入conf文件夹

cd conf

查看当前文件列表

修改zoo.cfg配置文件

# zookeeper 服务器心跳时间，单位为ms
tickTIme=2000
# 投票选举新Leader 的初始化时间
initLimit=10
# leader 与follower 心跳检测最大容忍时间，
# 响应超过syncLimit*ticketTime,leader 认为follower死掉，
#从服务器列表中删除follower
syncLimit=5
# 数据目录
dataDir=/tmp/zookeeper/data
# 日志目录
dataLogDir=/tmp/zookeeper/log
# 对外服务端口
clientPort=2181

默认情况下Linux中没有/tmp/zookeeper/data 和/tmp/zookeeper/log 这两个目录，因此我们需要手动创建数据文件夹

mkdir -p /tmp/zookeeper/data

创建日志文件夹

mkdir -p /tmp/zookeeper/log

在/tmp/zookeeper/data 文件夹下要创建一个myid 文件

vi myid

进入编辑模式

i

写入一个数字比如0

0

退出编辑模式，按ESC.
保存并退出

:wq

启动Zookeeper 服务器

zkServer.sh start

执行结果如下：

检查Zookeeper 执行状态

zkServer.sh status

执行结果如下：

我们可以看到，Mode 类型是standalone 而不是leader，也不是follower。
这是为什么呢？
那是因为当前我们配置的是Zookeeper 单个服务器而不是集群。
如果配置了多个Zookeeper 集群，则会出现一个leader，多个follower 。

多学一招

然后在三台机器的zoo.cfg 中配置

server.0=192.168.0.2:2888:3888
server.1=192.168.0.3:2888:3888 
server.2=192.168.0.4:2888:3888 

server.A=B:C:D
A是一个数字，代表服务器编号，就是myid 里面的值
B是服务器的IP地址
C服务器与集群中的leader服务器交换信息的端口
D:选举时服务器相互通信的端口

• Kafka 官网：http://kafka.apache.org/
• Kafka下载地址：http://kafka.apache.org/quickstart

复制到/opt 文件夹，然后执行解压命令

tar zxvf kafka_2.11-2.0.0.tgz

打开环境变量配置文件

vi /etc/profile

进入VI编辑器模式

i

添加环境变量

export KAFKA_HOME=/opt/kafka_2.11-2.0.0
export PATH=$PATH:$KAFKA_HOME/bin

修改完成后，退出编辑模式按键盘上的ESC
保存并退出输入命令

:wq

使用source让环境变量即时生效

source /etc/profile

修改Kafka 的配置文件 $Kserver.properties

# broker 的编号，如果集群中有多个broker,则每个broker 的编号要设置的不同
broker.id=0
# 对外提供的服务入口地址
listeners=PLAINTEXT://localhost:9092
# 存放消息的日志文件
log.dirs=/tmp/kafka-logs/
# kafka所需要的Zookeeper集群地址
zookeeper.connect=localhost:2181/kafka

如果是集群部署，那么每个服务器的broker.id 要不一样， listeners 配置参数也需要修改为与broker 对应的IP地址与域名之后各自启动服务器即可
注意，

• 代码中我们需要使用的是localhost:9092
• localhost:2181/kafka 是在安装有Kafka 服务器上操作管理Topic 需要用到的。
• localhost 生产环境请配置成外网可用的IP.

当前控制台启动

bin/kafka-server-start.sh config/server.properties

后台运行Kafka服务

bin/kafka-server-start.sh -daemon config/server.properties

jps -l

执行成功如下：

出现上面所示，说明Kafka 和Zookeeper 都运行正常

创建一个 topic 命令如下:

kafka-topics.sh --zookeeper localhost:2181/kafka --create --topic topic-demo --replication-factor 1 --partitions 1 Create topic "topic-demo"

本篇完~