kafka集群安装部署文档
一、基本环境
1、linux操作系统 2、jdk 1.7及以上环境
二、集群环境部署部署
1、zookeeper集群部署； 2、kafka集群部署。
三、zookeeper集群部署
1、检查配置JAVA环境。检验方法：执行java –version 2、解压 zookeeper到指定目录。 例如：/seaapp/zookeeper
3、生成配置文件，拷贝或重命名 zoo_sample.cfg文件 cp /seaapp/zookeeper /conf/zoo_sample.cfg zoo.cfg 4、修改配置文件 zoo.cfg vi zoo.cfg，修改 maxClientCnxns=300 #最大客户端连接数 dataDir=/seaapp /zookeeper/data server.1=1.2.3.4:2888:3888 server.2=1.2.3.5:2888:3888 server.3=1.2.3.6:2888:3888 要注意下server.1这个后缀，表示的是1.2.3.4这个机器，在机器中的server id是1。 dataDir是zookeeper数据文件和日志文件存放的目录。 5、创建数据目录(dataDir配置的目录)： mkdir /seaapp/zookeeper/data 6、：写入标识 Server ID.
在/seaapp /zookeeper/data目录中创建文件 myid 文件，每个文件中分别写入当前机器的 server id，例如1.2.3.4这个机器，在/seaapp/zookeeper/data目录的myid文件中写入数字1。 7、启动zookeeper：执行 /seaapp/zookeeper/bin/zkServer.sh start ##关闭zookeeper 执行 /seaapp/zookeeper/bin/zkServer.sh stop 8、检测是否成功启动：执行 1. /seaapp/zookeeper/bin/zkCli.sh 2. 或
3. echo stat|nc localhost 2181
四、kafka集群部署
1、解压kafka到指定目录。 例如：/seaapp/kafka
2、修改配置文件server.properties。
路径：/seaapp/kafka/config/server.properties vi server.properties 修改下列参数
broker.id： 每一个broker在集群中的唯一id，要求是正数
host.name: 这个host.name是Kafka的server的机器ip，会注册到Zookeeper中 log.dirs： kafka数据的存放地址（）
log.retention.hours：数据文件保留时间，单位h（根据log.dir目录空间和消息量来确定） zookeeper.connect：指定ZooKeeper的connect string，以hostname:port的形式，可有多个以逗号分隔，如hostname1:port1,hostname2:port2,hostname3:port3 auto.create.topics.enable=false 设置禁止自动创建topic
message.max.bytes=2048576 #服务端可接收消息的最大长度
replica.fetch.max.bytes=2048576 #备份同步获取最大长度,该值要不小于message.max.bytes
3、命令 启动： bin/kafka-server-start.sh config/server.properties &，生产环境进程在后台运行 结束： bin/kafka-server-stop.sh
五、kafka监控工具
kafka监控工具KafkaOffsetMonitor
程序所在目录：/seaapp/KafkaOffsetMonitor 启动脚本start.sh 启动：./start.sh 关闭：Ctrl + c
配置在脚本start.sh。主要修改zk的地址。port端口是监控页面的服务端口 java -cp KafkaOffsetMonitor-assembly-0.2.0.jar \\ com.quantifind.kafka.offsetapp.OffsetGetterWeb \\ --zk localhost:2188 \\ --port 80 \\
--refresh 10.seconds \\ --retain 1.days
Consumer Groups:
offset:当前groupid组中的消费者消费了的消息的位置
logsize： kafka服务器收到生产者发送到该主题的消息总数 Lag：未被该groupid组中消费者消费的消息数量。
六、kafka topic操作
kafka topic操作脚本位置：/seaapp/kafka/bin/ 脚本: kafka-topics.sh 参数：
--alter：更改topic 的partition 数量、replica assignment、configuration for the topic. --create： 指定创建topic动作 --topic：指定新建topic的名称
--zookeeper： 指定kafka连接zk的连接url，该值和server.properties文件中的配置项{zookeeper.connect}一样
--config：指定当前topic上有效的参数值，参数列表参考文档为: Topic-level configuration
--partitions：指定当前创建的kafka分区数量，默认为1个 --replication-factor：指定每个分区的复制因子个数，默认1个 --describe： 指定是展示详细信息命令 --list 查看topic列表
1、查看 topic列表信息
./kafka-topics.sh -zookeeper 127.0.0.1:2181 –list
2、查看topic的详细信息
./kafka-topics.sh –zookeeper 127.0.0.1:2181 -describe -topic ANALOG_DATA_GUODIAO
Topic:ANALOG_DATA_GUODIAO PartitionCount:2 ReplicationFactor:1 Configs: Topic: ANALOG_DATA_GUODIAO Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Topic: ANALOG_DATA_GUODIAO Partition: 1 Leader: 0 Replicas: 0 Isr: 0
3、为topic增加partition
./bin/kafka-topics.sh –zookeeper127.0.0.1:2181 –alter –partitions 2 –topic testKJ1
七、常见问题处理方法
1、kafka数据目录更换或想清空kafka现有数据和主题 ①停止所有kafka节点；
②停止所有Zookeeper节点；
③清空kafka数据目录（log.dirs配置的目录）文件；或重新配置log.dir参数； ④清空Zookeeper数据目录（dataDir）下的version-2的数据。 ⑤重启Zookeeper节点； ⑥重启kafka节点； ⑦创建kafka 的主题； ⑧重启和kafka相关的程序
2、如何清除consumer(group.id=test)的历史offset值？ ①停止consumer所在的程序；
②启动Zookeeper的zkCli.sh脚本，在/seaapp/zookeeper/bin/目录下。
③执行命令: ls /consumers 查看是否存在test的目录，如果存在，表示存在历史offset值 ④执行删除目录命令： rmr /consumers/test
⑤重启consumer所在的程序，consumer将从设置的方式开始读取kafka中的消息。
3、关于kafka中消息长度的问题
在第四节kafka集群配置的配置文件有有两个参数，message.max.bytes 、replica.fetch.max.bytes。这两个参数的默认值是1M，即一个kafka消息的最大长度为1M。kafka服务端接收的消息最大值为1M。若kakfa服务器的这两个参数修改，则应该修改kafka的producer和consumer的参数，不然的话会发生消息发送不出去或者消息太大，consumer接收不到消息的错误。
kafka producer的参数应该设置max.request.size 的参数，该参数的值不能大于服务端配置的
message.max.bytes的值。
kafka consumer的参数应该设置fetch.message.max.bytes的值，该参数的值应该不小于服务端配置的message.max.bytes的值。
