kafka集群/单节点部署

一、环境说明

系统：ubuntu1604，环境：java8

二、前戏

kafka操作目录是/usr/local/kafka，因此要先在/usr/local下新建kafka目录

三、zookeeper安装

进入kafka目录，下载最新的stable版本，命令： 

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz

解压到当前目录下，设置环境，命令：

vi /etc/profile

在最后加上这行：

export zookeeper_home=/usr/local/kafka/zookeeper-3.4.10

 

使它立即生效，命令：

source /etc/profile

进入/usr/local/kafka/zookeeper-3.4.10/bin下，启动zookeepeer，命令：

./zkServer.sh start

观察到Starting zookeeper ... STARTED就为成功

四、kafka安装

进入到/usr/local/kafka/目录下，下载kafka，命令：

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/0.10.2.0/kafka_2.11-0.10.2.0.tgz

 修改zookeeper.properties：

解压到当前目录下，进入/usr/local/kafka/kafka_2.11-0.10.2.0/config/，修改zookeeper.properties文件。其中server.1,server.2,server.3根据部署的节点数量和ip调整，例如是3个节点，ip分别为A：192.168.0.33，B：192.168.0.34，C：192.168.0.35，则修改的配置如下：

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/var/zookeeper
dataLogDir=/var/log/zookeeper
clientPort=2181
maxClientCnxns=0
server.1=192.168.0.33:2888:3888
server.2=192.168.0.34:2888:3888
server.3=192.168.0.35:2888:3888

修改server.properties

  进入/usr/local/kafka/kafka_2.11-0.10.2.0/config/，修改server.properties文件。其中broker.id配置A机器上面为0，B机器上面为1，C机器上面为2，端口号都保持默认，则A机器配置如下：

broker.id=0
delete.topic.enable=true
listeners=PLAINTEXT://192.168.0.34:9092
log.dirs=/var/log/kafka/kafka-logs
zookeeper.connect=192.168.0.34:2181,192.168.0.33:2181,192.168.0.35:2181

五、启动前检测

先启动zookeeper服务，命令：

/usr/local/kafka/kafka_2.11-0.10.2.0/bin/zookeeper-server-start.sh /usr/local/kafka/kafka_2.11-0.10.2.0/config/zookeeper.properties

观察如果成功一般都是info打印，再开启一个终端启动kafka，命令：

/usr/local/kafka/kafka_2.11-0.10.2.0/bin/kafka-server-start.sh /usr/local/kafka/kafka_2.11-0.10.2.0/config/server.properties

观察如果启动成功都是info打印，再开启一个终端进入/usr/local/kafka/kafka_2.11-0.10.2.0，创建topic，注意--replication-factor 后面的参数根据机器数量调整，本例是3，命令：

bin/kafka-topics.sh --create --zookeeper 192.168.0.34:2181 --replication-factor 3 --partitions 1 --topic test

再开启一个终端进入/usr/local/kafka/kafka_2.11-0.10.2.0启动consumer，命令：

bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.34:9092 --from-beginning --topic test

再开启一个终端进入/usr/local/kafka/kafka_2.11-0.10.2.0启动producer，命令：

bin/kafka-console-producer.sh --broker-list 192.168.0.34:9092 --topic test

在producer上输入任意内容，在consumer上可以看到相应的打印

由于是from begining，所以会把所有的都打印出来

六、托管给supervisor

安装supervisor，命令：

easy_install supervisor

进入/etc/supervisor/conf.d目录下，创建zookeeper配置文件，文件名字为zookeeper.conf，内容：

[program:zookeeper]
command=/usr/local/kafka/kafka_2.11-0.10.2.0/bin/zookeeper-server-start.sh /usr/local/kafka/kafka_2.11-0.10.2.0/config/zookeeper.properties
user=root
autostart=true
autorestart=true
startsecs=3

更新到supervisor，命令：

supervisorctl update

进入/etc/supervisor/conf.d目录下，创建kafka配置文件，文件名字为kafka.conf，内容：

[program:kafka-server]
command=/usr/local/kafka/kafka_2.11-0.10.2.0/bin/kafka-server-start.sh /usr/local/kafka/kafka_2.11-0.10.2.0/config/server.properties
user=root
autostart=true
autorestart=true
startsecs=3

更新到supervisor，命令：

supervisorctl update

此时进入supervisor，可以看到如下

大功告成，再按照上一步中的启动下consumer和producer

 七、kafka启动时遇到的问题

ERROR Processor got uncaught exception
大致错误如下，这段是别人的，我大概引用过来
java.lang.ExceptionInInitializerError
at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:110)
at kafka.network.Processor$$anonfun$processCompletedReceives$1.apply(SocketServer.scala:488)
at kafka.network.Processor$$anonfun$processCompletedReceives$1.apply(SocketServer.scala:483)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at kafka.network.Processor.processCompletedReceives(SocketServer.scala:483)
at kafka.network.Processor.run(SocketServer.scala:413)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.UnknownHostException: E-TEST-1: E-TEST-1: Name or service not known
at java.net.InetAddress.getLocalHost(InetAddress.java:1473)
at kafka.network.RequestChannel$.<init>(RequestChannel.scala:38)
at kafka.network.RequestChannel$.<clinit>(RequestChannel.scala)
... 10 more
Caused by: java.net.UnknownHostException: E-TEST-1: Name or service not known
at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method)
at java.net.InetAddress$1.lookupAllHostAddr(InetAddress.java:901)
at java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1293)
at java.net.InetAddress.getLocalHost(InetAddress.java:1469)
... 12 more
[2016-10-08 11:41:05,242] ERROR Processor got uncaught exception. (kafka.network.Processor)
[2016-10-08 17:22:47,558] WARN Could not initialize class kafka.network.RequestChannel$ (kafka.utils.CoreUtils$)
java.lang.NoClassDefFoundError: Could not initialize class kafka.network.RequestChannel$
at kafka.server.KafkaRequestHandler.shutdown(KafkaRequestHandler.scala:67)
at kafka.server.KafkaRequestHandlerPool$$anonfun$shutdown$2.apply(KafkaRequestHandler.scala:90)
at kafka.server.KafkaRequestHandlerPool$$anonfun$shutdown$2.apply(KafkaRequestHandler.scala:89)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
at kafka.server.KafkaRequestHandlerPool.shutdown(KafkaRequestHandler.scala:89)
at kafka.server.KafkaServer$$anonfun$shutdown$3.apply$mcV$sp(KafkaServer.scala:549)
at kafka.utils.CoreUtils$.swallow(CoreUtils.scala:76)
at kafka.utils.Logging$class.swallowWarn(Logging.scala:92)
at kafka.utils.CoreUtils$.swallowWarn(CoreUtils.scala:47)
at kafka.utils.Logging$class.swallow(Logging.scala:94)
at kafka.utils.CoreUtils$.swallow(CoreUtils.scala:47)
at kafka.server.KafkaServer.shutdown(KafkaServer.scala:549)
at kafka.server.KafkaServerStartable.shutdown(KafkaServerStartable.scala:49)
at kafka.Kafka$$anon$1.run(Kafka.scala:63)
注意这行
Caused by: java.net.UnknownHostException: E-TEST-1: Name or service not known
因为我的这台服务器名字是：E-TEST-1，所以解决办法是，进入/etc/hosts文件添加192.168.0.34 E-TEST-1这一行即可解决