2.kafka安装

这篇文章主要讲解kafka单机安装，以及集群安装。kafka的单机安装非常简单，只需按照下一段落提到的几步操作即可。

单机安装

1.下载

首先从官方下载安装包，官方地址：。kafka安装包和一般安装包的命名方式不一样，我们看一个kafka包命名：，其中2.11是scala的版本，1.1.0才是kafka的版本。官方强烈建议scala版本和服务器上的scala版本保持一致，避免引发一些不可预知的问题。官方原文如下：

We build for multiple versions of Scala. This only matters if you are using Scala and you want a version built for the same Scala version you use. Otherwise any version should work (2.11 is recommended).

2.解压

将安装包放到服务器上某个位置，执行如下命令：

3.启动zk

kafka依赖zookeeper，zookeeper集群可以自己搭建，也可以用kafka安装包中内置的shell脚本启动zookeeper。

如果已经安装了zk单机或者zk集群（单机或者集群都可以，取决于你对高可用的要求），那么可以跳过这一步：

这种启动方式如果执行ctrl+z后会退出，启动的进程也会退出，所以建议增加参数启动：

说明：笔者后面的测试，假设在10.0.55.209上部署了独立zk，而不是通过kafka安装包中的脚本部署zk，且端口为默认端口2181。

4.配置

启动kafka之前，需要确认一些配置信息，kafka的配置信息在config/server.properties文件中，笔者配置kafka直接显示指定绑定的ip为，所以需要增加如下一下配置：

说明：zookeeper.connect配置可以附上chroot路径，例如，推荐这样配置。即使多个kafka集群共享zk集群也不会有冲突。另外，如果zk配置集群的话，只需多个地址用逗号隔开即可，比如：10.0.55.208:2181/myzk,10.0.55.209:2181/myzk,10.0.55.210:2181/myzk。

5.启动

执行如下脚本即可：

这种启动方式如果执行ctrl+z后会退出，启动的进程也会退出，所以建议增加参数启动：

-daemon运行原理：

为什么启动kafka时增加参数就能实现后台进程方式启动呢？我们首先看一下最后一行：，其本质是调用了，这个脚本的部分代码如下，DAEMON_MODE为true，就能通过方式启动，而DAEMON_MODE为true的前提是COMMAND 为-daemon，COMMAND就是第一个参数：，所以启动脚本增加就能实现后台进程方式启动：

6.发送消息

发送消息之前，先创建topic（kafka能自动创建topic，通过配置决定，但是一般生产环境建议关闭该特性）：

使用kafka提供的发送消息：

7.接收消息

使用kafka提供的接收消息：

集群安装

单机版kafka安装并可以成功发送&接收消息后，搭建kafka集群就比较简单了。

1.配置文件

首先基于conf目录下的server.propreties文件，复制两个文件server-1.propreties，server-2.propreties

server-1.propreties做如下变更：

server-2.properties做类似的变更：

由于笔者将三个kafka broker放在一台linux服务器上，所以不能一样（如果每个broker一台服务器，那么port可以一致，生产环节肯定是每个broker部署在一台独立的服务器上）。但是无论三个broker是否在一台服务器上，一定是整个集群下全局唯一。否则无法正常启动broker。其原理是在zk上以broker.id的值写入对应的节点，如果写入失败，broker启动就失败。所以整个集群的broker.id要保证全局唯一。

2.启动kafka

执行如下两个命令，分别再启动两个kafka broker：

3.集群验证

由于总计部署了3个broker，所以创建topic时能指定，如果不能指定，说明集群部署有问题：

查看刚才创建的topic详情，可以看到总计5个分区（PartitionCount），且副本数为3（ReplicationFactor），且每个分区上有3个副本（通过Replicas的值可以得出），另外最后一列Isr（In-Sync Replicas）即表示处理同步状态的副本集合，这些副本与leader副本保持同步，没有任何同步延迟，另外Leader，Replicas Isr中的0，1，2就是broker id，对应配置文件conf/server.properties中的：