前面刚和大家讲过Kafka的单机是如何安装的，但是对于大数据来说，是需要搭建多个Kafka Brokers。我们在配置server.properties中已有一个Kafka代理实例。 现在我们需要多个代理实例，因此将现有的server.properties文件拷贝到另外2台机的配置文件中，在这里为了方便起见，就直接拷贝有配置文件的文件夹kafka211到另外2台机上,然后直接改配置文件信息就好了，省了另外两台机安装kafka的时间。

kafka是依赖zookeeper的，前面已经说过，而zookeeper集群的搭建也已经给大家说过，还没有搭建好的朋友可以再去看看
详细zookeeper分布式集群的搭建
今天主要带大家搭建kafka集群。
还是使用和前面创建时候一样，我们还是用3台机来搭建，就用前面搭建spark集群时候的步骤一样，先用spark集群的主机名；

还是分别为每台虚拟机设置过的主机名：
hostname spark1
hostname spark2
hostname spark3

1.修改zookeeper.properties

前面说过配置zookeeper.properties 如果是单击就不用配置，这里是搭建集群的关系，所以要开始配置这个文件，是zookeeper日志的存放目录,注意日志目录是你之前配置zookeeper时在conf写的目录

cd /opt/soft/kafka211/config
vi zookeeper.properties 

修改为以下目录

dataDir=/opt/soft/zk345/tmp  clientPort=2181

2.修改server.properties

vi server.properties
主要是要改下面4个地方的配置

• spark1这台机的broker.id已经配置过了，就不用再改，只需要改另外两台机的broker.id，使3台机的值不一样就行，可以分别配置为0，1，2
• kafka端口号默认是9092,可以不用修改，也可以为每台机改一个新端口号，比如spark1 配置port=9092 ; spark2 配置port=9093 ;spark3 配置port=9094
• log.dirs地址都是一样的也不用在修改
• 主要是配置zookeeper.connect,要把3台机的IP都配置在一起

zookeeper.connect=spark1:2181,spark2:2181,spark3:2181
或者把主机名改成每台机的ip地址都行

也可以多添加一个配置delete.topic.enable=true主要是用来删除topic
环境变量已经配置过了，在这里就不多说了，只需要在两外两台机上取配置并激活就好了。
把已经重新配置过的文件拷贝到另外两台机上

3.拷贝配置文件

scp -r /opt/soft/kafka211 root@spark2:$PWD
scp -r /opt/soft/kafka211 root@spark3:$PWD

如果另外两台机都出现kafka211这个文件夹说明拷贝成功，这时另外两台机都要配置kafka环境变量，拷贝后如果出现访问权限不够的报错情况，前面在讲spark集群的时候有说过结局方案，可以再去看看。

source /etc/profile#添加以下环境变量
export KAFKA_HOME=/opt/soft/kafka211
export PATH=$PATH:$KAFKA_HOME/bin

激活配置

source /etc/profile

修改另外两台机的server.properties配置 文件，改成
spark2的broker.id=1
spark3的broker.id=2

4.启动zookeeper集群

这里就不多说了，每台一个个单独启动

zkServer.sh start

jps查看进程，如果有QuorumPeerMain字样说明集群已经启动成功！

5.启动kafka集群

在bin目录启动下：

cd /opt/soft/kafak211/bin
./kafka-server-start.sh /opt/soft/kafka211/config/server.properties 

另外两台机也是按照这个步骤和命令启动kafka集群，直到进程中出现kafak的字样。
三台机均要启动；启动无报错，即搭建成功，可以生产和消费消息，来检测是否搭建成功。

创建Topic来验证集群是否创建成功

./kafka-topics.sh --create --zookeeper 192.168.5.150:12181 --replication-factor 2 --partitions 1 --topic shuaige

#解释
–replication-factor 2 #复制两份
–partitions 1 #创建1个分区
–topic #主题为shuaige

在一台服务器上创建一个发布者
创建一个broker，发布者

./kafka-console-producer.sh --broker-list 192.168.5.150:9092 --topic shuaige

在一台服务器上创建一个订阅者

./kafka-console-consumer.sh --bootstrap-server 192.168.5.150:9092 --topic shuaige --from-beginning

测试（在发布者那里发布消息看看订阅者那里是否能正常收到~）：

配置文件server.properties参数解析

• broker.id=0
  #当前机器在集群中的唯一标识，和zookeeper的myid性质一样

• port=9092
  #当前kafka对外提供服务的端口默认是9092

• host.name=192.168.5.150
  #这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。

• num.network.threads=3
  #这个是borker进行网络处理的线程数

• num.io.threads=8
  #这个是borker进行I/O处理的线程数

• log.dirs=/opt/soft/kafka211/datas
  #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个

• socket.send.buffer.bytes=102400
  #发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能

• socket.receive.buffer.bytes=102400
  #kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘

• socket.request.max.bytes=104857600
  #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小

• num.partitions=1
  #默认的分区数，一个topic默认1个分区数

• log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天

• message.max.byte=5242880 #消息保存的最大值5M

• default.replication.factor=2 #kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务

• replica.fetch.max.bytes=5242880 #取消息的最大直接数

• log.segment.bytes=1073741824 #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件
  log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间（log.retention.hours=168 ），到目录查看是否有过期的消息如果有，删除

• log.cleaner.enable=false #是否启用log压缩，一般不用启用，启用的话可以提高性能

• zookeeper.connect=192.168.5.150:2181 #设置zookeeper的连接端口