Kafka安装配置及使用说明.doc

Kafka安装配置及使用说明

（铁树2018-08-08）

（Windows平台，5个分布式节点，修改消息大小，调用程序范例）

1 安装配置
采用5台服务器作为集群节点，IP地址为：XX.XX.0.12-XX.XX.0.16.
每台机器依次安装配置JDK、zookeeper、kafka，先安装完一台机器，然后拷贝到其他机器，再修改配置文件。
1.1 JDK安装配置
JDK版本：jdk1.7.0_51_x64解压版（jdk1.7.0_51_x64.rar）
解压到C盘kafka目录下，如图所示。

设置环境变量：
JAVA_HOME：C:\kafka\jdk1.7.0_51_x64
PATH：C:\kafka\jdk1.7.0_51_x64\bin

1.2 zookeeper安装配置
1.2.1 解压安装
zookeeper版本：3.4.12 （zookeeper-3.4.12.tar.gz）
解压到C盘kafka目录下，如图所示。

1.2.2 创建zookeeper数据目录和日志目录
zkdata #存放快照
C:\kafka\zookeeper-3.4.12\zkdata
zkdatalog#存放日志
C:\kafka\zookeeper-3.4.12\zkdatalog
1.2.3 修改配置文件
进入到“C:\kafka\zookeeper-3.4.12”目录下的conf目录中，复制zoo_sample.cfg（官方提供的zookeeper的样板文件），重命名为zoo.cfg（官方指定的文件命名规则）。

默认内容：

修改后配置文件为：

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=C:/kafka/zookeeper-3.4.12/zkdata
dataLogDir=C:/kafka/zookeeper-3.4.12/zkdatalog
# the port at which the clients will connect
clientPort=12181
server.1=XX.XX.0.12:12888:13888
server.2=XX.XX.0.13:12888:13888
server.3=XX.XX.0.14:12888:13888
server.4=XX.XX.0.15:12888:13888
server.5=XX.XX.0.16:12888:13888

# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
autopurge.snapRetainCount=100
# Purge task interval in hours
# Set to "0" to disable auto purge feature
autopurge.purgeInterval=24

配置文件解释：
#tickTime：
这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。
#initLimit：
这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 10个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 10*2000=20 秒
#syncLimit：
这个配置项标识 Leader 与Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是5*2000=10秒
#dataDir：
快照日志的存储路径
#dataLogDir：
事物日志的存储路径，如果不配置这个那么事物日志会默认存储到dataDir制定的目录，这样会严重影响zk的性能，当zk吞吐量较大的时候，产生的事物日志、快照日志太多
#clientPort：
这个端口就是客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求。修改他的端口改大点
通过配置 autopurge.snapRetainCount 和 autopurge.purgeInterval 这两个参数能够实现定时清理了。这两个参数都是在zoo.cfg中配置的：
autopurge.purgeInterval  这个参数指定了清理频率，单位是小时，需要填写一个1或更大的整数，默认是0，表示不开启自己清理功能。
autopurge.snapRetainCount 这个参数和上面的参数搭配使用，这个参数指定了需要保留的文件数目。默认是保留3个。

1.2.4 创建myid文件
在“C:\kafka\zookeeper-3.4.12\zkdata”目录下，创建myid文件（无后缀名），内容为对应IP地址的主机号。如server.1则内容为1。

1.3 Kafka安装配置
1.3.1 解压安装
kafka 版本：kafka1.1.1（kafka_2.11-1.1.1.tgz）
解压到C盘kafka目录下，如图所示。

1.3.2 创建消息目录
kafkalogs ：C:\kafka\kafka_2.11-1.1.1\kafkalogs

1.3.3 修改配置文件
打开C:\kafka\kafka_2.11-1.1.1\config\ server.properties 

实际的修改项为：
broker.id=1
listeners=PLAINTEXT://:19092
log.dirs= C:/kafka/kafka_2.11-1.1.1/kafkalogs
#在log.retention.hours=168 下面新增下面三项（消息大小最大1GB）
message.max.byte=1073741824
replica.fetch.max.bytes=1073741824
log.segment.bytes=1073741824
default.replication.factor=2

#设置zookeeper的连接端口
zookeeper.connect=XX.XX.0.12:12181,XX.XX.0.13:12181,XX.XX.0.14:12181,XX.XX.0.15:12181,XX.XX.0.16:12181

配置说明：
broker.id=0 #当前机器在集群中的唯一标识，和zookeeper的myid性质一样
port=19092 #当前kafka对外提供服务的端口默认是9092
host.name=192.168.7.100 #这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。
work.threads=3 #这个是borker进行网络处理的线程数
num.io.threads=8 #这个是borker进行I/O处理的线程数
log.dirs=/opt/kafka/kafkalogs/ #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的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.7.100:12181,192.168.7.101:12181,192.168.7.107:1218 #设置zookeeper的连接端口

1.4 其他节点配置
将安装以上配置好的目录c:\kafka拷贝到其他节点的c盘目录，并修改如下配置。
1、JAVA环境变量：
JAVA_HOME：C:\kafka\jdk1.7.0_51_x64
PATH：C:\kafka\jdk1.7.0_51_x64\bin
2、zookeeper的myid
C:\kafka\zookeeper-3.4.12\zkdata\myid，修改为对应的数值
XX.XX.0.12：1
XX.XX.0.13：2
XX.XX.0.14：3
XX.XX.0.15：4
XX.XX.0.16：5
3、kafka配置
C:\kafka\kafka_2.11-1.1.1\config\ server.properties的
broker.id，修改为对应的数值
XX.XX.0.12：1
XX.XX.0.13：2
XX.XX.0.14：3
XX.XX.0.15：4
XX.XX.0.16：5

1.5 服务启动
1、 启动zookeeper
C:\kafka\zookeeper-3.4.12\bin\zkServer.cmd
XX.XX.0.12-16，依次双击启动。

2、 启动kafka
运行cmd，cd C:\kafka\kafka_2.11-1.1.1目录，再执行命令：
【cd C:\kafka\kafka_2.11-1.1.1】
C:\kafka\kafka_2.11-1.1.1>.\bin\windows\kafka-server-start.bat .\config\server.properties

1.6 服务状态测试
1.6.1 创建Topics
打开cmd 进入C:\kafka\kafka_2.11-1.1.1\bin\windows
C:\kafka\kafka_2.11-1.1.1\bin\windows>kafka-topics.bat --create --zookeeper localhost:12181 --replication-factor 1 --partitions 1 --topic test001

1.6.2 打开一个Producer
打开cmd 进入C:\kafka\kafka_2.11-1.1.1\bin\windows
C:\kafka\kafka_2.11-1.1.1\bin\windows>kafka-console-producer.bat --broker-list localhost:19092 --topic test001
>等待输入消息内容。

1.6.3 打开一个Consumer
打开cmd 进入C:\kafka\kafka_2.11-1.1.1\bin\windows
C:\kafka\kafka_2.11-1.1.1\bin\windows>kafka-console-consumer.bat --zookeeper localhost:12181 --topic test001

然后就可以在Producer控制台窗口输入消息了，很快Consumer窗口就会显示出Producer发送的消息。

1.6.4 查看所有主题
C:\Users\Develop>C:\kafka\kafka_2.11-1.1.1\bin\windows\kafka-topics.bat --list --zookeeper localhost:12181

1.6.5 查看Topic分区和副本
C:\Users\Develop>C:\kafka\kafka_2.11-1.1.1\bin\windows\kafka-topics.bat --describe --zookeeper localhost:12181

1.7 消息大小调整
Kafka对于10KB大小的消息吞吐率最好，默认配置最大支持1MB的消息大小。对于大消息的传输，需要修改kafka的server.properties、consumer、producer的相关配置。
server.properties修改：
打开C:\kafka\kafka_2.11-1.1.1\config\ server.properties
（按照最大1GB）
message.max.bytes=1073741824
replica.fetch.max.bytes=1073741824
log.segment.bytes=1073741824

consumer配置：
max.partition.fetch.bytes=1073741824

Producer配置：
max.request.size =1073741824
#33554432，默认32M
buffer.memory= 1073741824

mon.errors.RecordTooLargeException: The message is 36428062 bytes when serialized which is larger than the total memory buffer you have configured with the buffer.memory configuration.
附件太大可能会内存溢出，还会涉及超时参数配置等。

2 JAVA程序示例
2.1 Producer程序示例
2.1.1 Properties文件配置
##producer
bootstrap.servers=XX.XX.0.12:19092,XX.XX.0.13:19092,XX.XX.0.14:19092,XX.XX.0.15:19092,XX.XX.0.16:19092
producer.type=sync
request.required.acks=1

##consumer
mit=true
#latest, earliest, none
auto.offset.reset=earliest

建议公共参数（如服务地址）配置在properties文件里。其他参数根据接口需要程序中配置。
// 创建Producer
private Producer<Integer, String> createProducer() {
Properties props = new Properties();
String path = ProducerDemo.class.getResource("/").getFile().toString()+ "kafka.properties";
try {
FileInputStream fis = new FileInputStream(new File(path));
props.load(fis);
props.put("key.serializer", "mon.serialization.IntegerSerializer");
props.put("value.serializer","mon.serialization.StringSerializer");
fis.close();
} catch (Exception e) {
e.printStackTrace();
}
return new KafkaProducer<Integer, String>(props);
}
2.1.2 Properties配置详解
# 0:producer不会等待broker发送ack
# 1:当leader接收到消息后发送ack
# all(-1):当所有的follower都同步消息成功后发送ack
request.required.acks=0

2.1.3 主题+VALUE

import java.io.File;
import java.io.FileInputStream;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class TopicValue {
// 创建Producer
private Producer<String, String> createProducer() {
Properties props = new Properties();
String path = ProducerDemo.class.getResource("/").getFile().toString()
+ "kafka.properties";
try {
FileInputStream fis = new FileInputStream(new File(path));
props.load(fis);
props.put("key.serializer", "mon.serialization.StringSerializer");
props.put("value.serializer","mon.serialization.StringSerializer");
fis.close();
} catch (Exception e) {
e.printStackTrace();
}
return new KafkaProducer<String, String>(props);
}

public static void main(String[] args) {
// 消息主题
String topicName="test001";
TopicValue topicValueProducer=new TopicValue();
Producer<String, String> producer = topicValueProducer.createProducer();
producer.send(new ProducerRecord<String, String>(topicName, "消息: TopicValue"));
producer.flush();
producer.close();
System.out.println("Message send successfully");
}
}

2.1.4 主题+KEY+VALUE
2.1.4.1 <Integer,String>
package kjsp.kafka.producer;

import java.io.File;
import java.io.FileInputStream;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class TopicIntegerString {
// 创建Producer
private Producer<Integer, String> createProducer() {
Properties props = new Properties();
String path = ProducerDemo.class.getResource("/").getFile().toString()
+ "kafka.properties";
try {
FileInputStream fis = new FileInputStream(new File(path));
props.load(fis);
props.put("key.serializer", "mon.serialization.IntegerSerializer");
props.put("value.serializer","mon.serialization.StringSerializer");
fis.close();
} catch (Exception e) {
e.printStackTrace();
}
return new KafkaProducer<Integer, String>(props);
}

public static void main(String[] args) {
// 消息主题
String topicName="test001";
TopicIntegerString topicValueProducer=new TopicIntegerString();
Producer<Integer, String> producer = topicValueProducer.createProducer();
producer.send(new ProducerRecord<Integer, String>(topicName, 1,"消息: TopicIntegerString1"));
producer.flush();
producer.close();
System.out.println("Message send successfully");
}
}

2.1.4.2 <String,String>
import java.io.File;
import java.io.FileInputStream;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class TopicStringString {
// 创建Producer
private Producer<String, String> createProducer() {
Properties props = new Properties();
String path = ProducerDemo.class.getResource("/").getFile().toString()
+ "kafka.properties";
try {
FileInputStream fis = new FileInputStream(new File(path));
props.load(fis);
props.put("key.serializer", "mon.serialization.StringSerializer");
props.put("value.serializer","mon.serialization.StringSerializer");
fis.close();
} catch (Exception e) {
e.printStackTrace();
}
return new KafkaProducer<String, String>(props);
}

public static void main(String[] args) {
// 消息主题
String topicName="test001";
TopicStringString topicValueProducer=new TopicStringString();
Producer<String, String> producer = topicValueProducer.createProducer();
producer.send(new ProducerRecord<String, String>(topicName, "TopicStringString001", "消息: TopicStringString001"));
producer.flush();
producer.close();
System.out.println("Message send successfully");
}
}
2.1.4.3 <String,byte[]>
package kjsp.kafka.producer;

import java.io.File;
import java.io.FileInputStream;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class TopicStringByte {
// 创建Producer
private Producer<String, byte[]> createProducer() {
Properties props = new Properties();
String path = ProducerDemo.class.getResource("/").getFile().toString()
+ "kafka.properties";
try {
FileInputStream fis = new FileInputStream(new File(path));
props.load(fis);
props.put("key.serializer", "mon.serialization.StringSerializer");
props.put("value.serializer","mon.serialization.ByteArraySerializer");
fis.close();
} catch (Exception e) {
e.printStackTrace();
}
return new KafkaProducer<String, byte[]>(props);
}

public static void main(String[] args) {
// 消息主题
String topicName="test001";
TopicStringByte topicValueProducer=new TopicStringByte();
Producer<String, byte[]> producer = topicValueProducer.createProducer();
producer.send(new ProducerRecord<String, byte[]>(topicName, "TopicStringByte001", "消息: TopicStringByte001".getBytes()));
producer.flush();
producer.close();
System.out.println("Message send successfully");
}
}
2.1.4.4 <byte[],byte[]>
package kjsp.kafka.producer;

import java.io.File;
import java.io.FileInputStream;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class TopicByteByte {
// 创建Producer
private Producer<byte[], byte[]> createProducer() {
Properties props = new Properties();
String path = ProducerDemo.class.getResource("/").getFile().toString()
+ "kafka.properties";
try {
FileInputStream fis = new FileInputStream(new File(path));
props.load(fis);
props.put("key.serializer", "mon.serialization.ByteArraySerializer");
props.put("value.serializer","mon.serialization.ByteArraySerializer");
fis.close();
} cat