本篇主要讲Linux下的Kafka环境搭建步骤，不是基于docker容器安装，搭建测试的环境包括CentOS7.7、Unbuntu18.08，其中有涉及引用其他博客网上资料，若有不妥之处，请私信处理。
首先 简要介绍一下Kafka：一种高吞吐量的分布式发布订阅消息系统，java开发的mq中间件，通过磁盘I0进行消息的持久化，每秒可以支持百万级的消息吞吐量，支持集群拓展、使用主从节点协调机制具有高容错性。其使用场景如：运行服务产生的日志、网站浏览监控分析、运营监控数据指标，大数据的流处理平台构建等。
Kafka的角色包括broker、toptic、partition、offset、producer、consumer、consumer group：broker表示一个运行Kafka的服务器节点；toptic是表示消息的种类的逻辑概念；partition是消息存储的物理概念，一个partition对应磁盘上的一个log文件，一个toptic分成多个partition存储分区，分布在多个broker上；producer是消息的生产者，可指定发布的topic和parition；offset是长整型，唯一标识每一条消息追加存储到partition的log文件的偏移量；consumer是消费者，对生产者产生的消息进行消费，consumer group是消费者组，每一个consumer都隶属于一个consumer，如果不指定则为默认，消息会发送给每个consumer group（消息广播)，但是同一个consumer group内部的consumers只能基于负载均衡消费消息，每个consumer消费的消息不重复（消息单播）。如下为详细的架构图示：
切入正题，Kafka的环境部署，由于Kafka是基于java语言开发的，因此需要JDK支持，同时依赖zookeeper注册进行节点管理，此外还有Kafka的下载，下面为下载链接：
JDK:Java SE Development Kit 8 - Downloads
zookeeper：zookeeper
kafka：kafka
安装流程是JDK->zookeeper->kafka，关于JDK的安装配置这里就不多说明了，网上资料搜索对于涉及过java的来说是比较简单的，接下来说说zookeeper和kafka：
对于zookeeper：
1.下载安装包之后使用linux的解压tar命令解压，如tar -zxvf apache-zookeeper-3.5.5-bin.tar.gz；
2.解压之后cd进入目录 cd apache-zookeeper-3.5.5-bin/config；
3.重命名zoo_sample.cfg文件：mv zoo_sample.cfg zoo.cfg；
对于kafka：
1.解压：tar -xzf kafka_2.12-2.3.0.tgz；
2.进入kafka目录：cd kafka_2.12-2.3.0/
完成以上流程只是进行下面的安装包进行解压和文件命名而已，后续配置，基于不同的模式需要进行不同的配置，对于解压后的zookeeper和kafka，配置中涉及两个修改项的目录，zookeeper解压后的子目录bin和conf，bin存放启动的程序，conf目录下的zoo.cfg存放配置参数，同理，kafka解压后的子目录bin也是存成启动程序，config文件夹下的server.properties存放配置参数，部署模式分成根据节点的个数分为集群和单机两种，根据通讯网络分为内网和外网，如下：
（一）集群部署
以三个节点为例，则需要部署三个zookeeper和kafka在三个不同的机器上运行，每一个节点如下：
1.zoo.cfg存放参数配置信息，编辑zookeeper的conf目录下的zoo.cfg文件夹：vi zoo.cfg，可修改或者新增如下参数：
initLimit=10 #集群中leader服务器与follower服务器第一次连接最多次数（可修改）
syncLimit=5 # leader 与 follower 之间发送和应答时间，如果该follower 在设置的时间内不能与leader 进行通信，那么此 follower 将被视为不可用。（可修改）
dataDir=/usr/local/zookeeper-3.4.6/data #自定义的zookeeper保存数据的目录（可修改）
clientPort=2181 #客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求，（可修改）
server.1=192.168.200.14:2888:38888#服务器编号=服务器IP:LF数据同步端口:LF选举端口（新增）
server.2=192.168.200.10:2888:3888（新增）
server.3=192.168.200.11:2888:3888（新增）
# 其中server配置对于集群中对应多少个节点就配置多少个，IP填写对应的服务器IP，两个端口一般为以上默认端口
2.配置zookeeper集群节点的ID：在每一个节点的zookeeper下的conf目录的同一级目录新建一个data文件夹，新建一个myid文件，写入节点的ID编号，如 执行 echo“1”>>/usr/local/apache-zookeeper-3.5.5-bin/data/myid ,表示将1写入zookeeper下的data目录下的myid文件，注意此ID不能重复，而且需要与上一点中的server编号进行一一对应，每一个节点都需创建此文件，写入该节点的ID编号；
3.修改每个节点的kafka解压目录下的config目录下的server.properties文件，修改或新增如下：
broker.id=1 # 指定节点的ID号，不同节点ID必须不同，同时与zookeeper中国的myid文件中的ID对应
listeners=PLAINTEXT://192.168.200.14:9092 # 监听的ip和端口
zookeeper.connect=192.168.200.14:2181,192.168.200.10:2181,192.168.200.11:2181 #对于集群每个节点的和端口的集合描述符。
（二）单机部署
1.修改单机的kafka解压目录下的config目录下的server.properties文件，修改或新增如下：
listeners=PLAINTEXT://192.168.200.14:9092 # 监听的ip和端口
zookeeper.connect=192.168.200.14:2181 #对于单机IP和端口的描述符。
（三）内网通讯
对于生产和消费连接kafka在本机运行通讯，只需要确认kafka解压目录下的config目录下的server.properties文件中： listeners=PLAINTEXT://192.168.200.14:9092 # 监听的ip和端口 ，此ip配置为该节点机器的本地ip，比如locahost表示本机，
如果生产和消费者连接kafka是在内网运行通讯，则该ip修改为主机名（可自定义，比如kafkahost），配置为listeners=PLAINTEXT://kafkahost:9092，然后编辑/etc/hosts文件，新增 比如：192.168.200.14 kafkahost ，建立内网IP和主机名的映射关系。
（四）外网通讯
对于生产者和消费者连接kafka在外网通讯，如果该节点机器已配置外网IP，则在kafka解压目录下的config目录下的server.properties文件中修改listeners配置，新增advertised.listeners设置，如下：
#当告诉服务器监听了 0.0.0.0，意味着监听每一个可用的网络接口
listeners=PLAINTEXT://0.0.0.0:9092
#该配置是提供给消费者和生产者的，如果没有配置，则默认将listeners的值给消费者和生产者 ，此IP为外网IP
advertised.listeners=PLAINTEXT://59.69.101.206:9092
对于生产者连接kafka所在的节点的机器为内网IP，需要通过IP和端口映射才能访问到的，推荐使用主机名替换listeners中的IP，如：
listeners=PLAINTEXT://hostname:9092
advertised.listeners=PLAINTEXT://hostname:9092
命令生产和消费时也用主机名代替真实IP，不过是客户端机器的/etc/hosts配置hostname映射成内网IP，外网客户端配成映射外网IP。
（五）服务启动
开启zookeeper命令(备注：先进入zookeeper的bin目录)
./zkServer.sh start
关闭zookeeper命令(备注：先进入zookeeper的bin目录)
./zkServer.sh stop
查看zookeeper的启动状态(备注：先进入zookeeper的bin目录)
./zkServer.sh status
Kafka启动命令(备注：先进入kafka目录)
常规模式启动kafka
bin/kafka-server-start.sh config/server.properties
进程守护模式启动kafka
nohup bin/kafka-server-start.sh config/server.properties >/dev/null 2>&1 &
Kafka关闭命令(备注：先进入kafka目录)
bin/kafka-server-stop.sh
启动成功可查看kafka进程
ps -ef|grep kafka
（六）测试
1.zookeeper测试：
对于单节点使用上述查询zookeeper状态便可以，对于多节点的zookeeper测试，比如zookeeper解压在/usr/local/目录下：
在任意节点上执行如下，创建topic为test1，发送hello成功，注意IP为该节点机器的IP和端口：
/usr/local/apache-zookeeper-3.5.5-bin/bin/zkCli.sh -server 192.168.200.10:2181 create /test1 "hello"
在其他节点上获取topic为test1的消息，输入如下命令显示收到hello，注意IP为该节点机器的IP和端口：
/usr/local/apache-zookeeper-3.5.5-bin/bin/zkCli.sh -server 192.168.200.11:2181
2.kafka测试：
创建话题"test"：
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
列出话题列表，可以查看创建的话题：
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
发送信息：
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
启动消费者，可以看到收到了刚才生产者发送的消息：
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
以上命令是用连接的kafka为本机单节点为例，命令参数为locahost：9092，如果为集群也可以改为如下某个 集群配置192.168.200.14:2181,192.168.200.10:2181,192.168.200.11:2181 。
