CentOS Install kafka

192.168.190.152
192.168.190.153
192.168.190.154

# tar zxvf jdk-8u65-linux-x64.tar.gz
# mv jdk1.8.0_65 /usr/local/
# vi /etc/profile.d/jdk.sh 
JAVA_HOME=/usr/local/jdk1.8.0_65
export JAVA_HOME
PATH=$JAVA_HOME/bin:$PATH
export PATH

# source /etc/profile

# wget http://www-eu.apache.org/dist/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz

# tar zxvf zookeeper-3.4.9.tar.gz
# mv zookeeper-3.4.9 /usr/local/
# useradd zookeeper
# chown -R zookeeper.zookeeper /usr/local/zookeeper-3.4.9

# cd /usr/local/zookeeper-3.4.9/conf/
# cp zoo_sample.cfg zoo.cfg
# mkdir /etc/zookeeper
# ln -sf /usr/local/zookeeper-3.4.9/conf/zoo.cfg /etc/zookeeper/
# mkdir -p /var/lib/zookeeper
# chown -R zookeeper:zookeeper /var/lib/zookeeper

# 用来标识主机
#echo 1 > /var/lib/zookeeper/myid
#echo 2 > /var/lib/zookeeper/myid 
#echo 3 > /var/lib/zookeeper/myid

# vi /etc/zookeeper/zoo.cfg 
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/var/lib/zookeeper
clientPort=2181
maxClientCnxns=0
server.1=192.168.190.152:2888:3888
server.2=192.168.190.153:2888:3888
server.3=192.168.190.154:2888:3888

[root@localhost ~]# vi /etc/init.d/zookeeper 
#! /bin/sh
#
# chkconfig: 2345 90 10
# description: zookeeper daemon

. /etc/init.d/functions

# You will probably want to change only two following lines.
BASEDIR="/usr/local/zookeeper-3.4.9"
USER="zookeeper"

PROG="zookeeper"
CMD="bin/zkServer.sh"
RETVAL=0

start () {
echo -n $"Starting ${PROG}: "
runuser ${USER} -c "cd ${BASEDIR}
${CMD} start > /dev/null &"
echo
}

stop () {
echo -n $"Stopping ${PROG}: "
runuser ${USER} -c "cd ${BASEDIR}
${CMD} stop > /dev/null &"
echo
}

restart () {
stop
start
}


# See how we were called.
case "$1" in
start)
start
;;
stop)
stop
;;
restart)
restart
;;
*)
echo $"Usage: $0 {start|stop|status}"
RETVAL=2
;;
esac

exit $RETVAL

# chmod 755 /etc/init.d/zookeeper
# /etc/init.d/zookeeper start
# chkconfig zookeeper on

# ./zkCli.sh

# 查看当前节点列表
[zk: localhost:2181(CONNECTED) 6] ls /
[zookeeper]

# 创建节点
[zk: localhost:2181(CONNECTED) 7] create /test "test"
Created /test
[zk: localhost:2181(CONNECTED) 8] ls /
[zookeeper, test]

# 查看节点数据
[zk: localhost:2181(CONNECTED) 9] get /test
test
cZxid = 0x100000004
ctime = Thu Oct 13 08:35:04 CST 2016
mZxid = 0x100000004
mtime = Thu Oct 13 08:35:04 CST 2016
pZxid = 0x100000004
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0

# 设置节点数据
[zk: localhost:2181(CONNECTED) 10] set /test "111111"
cZxid = 0x100000004
ctime = Thu Oct 13 08:35:04 CST 2016
mZxid = 0x100000005
mtime = Thu Oct 13 08:35:41 CST 2016
pZxid = 0x100000004
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0

# 删除节点数据
[zk: localhost:2181(CONNECTED) 11] delete /test
[zk: localhost:2181(CONNECTED) 12] ls /
[zookeeper]

192.168.190.152
192.168.190.153
192.168.190.154

wget http://mirrors.cnnic.cn/apache/kafka/0.10.0.1/kafka_2.11-0.10.0.1.tgz

# tar zxvf kafka_2.11-0.10.0.1.tgz
# mv kafka_2.11-0.10.0.1 /usr/local/
# cd /usr/local/kafka_2.11-0.10.0.1/

# vi config/server.properties 
broker.id=0 # id唯一 0 1 2
port=9092
host.name=192.168.190.152 #192.168.190.153 192.168.190.154
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
default.replication.factor=2
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.190.152:2181,192.168.190.153:2181,192.168.190.154:2181
zookeeper.connection.timeout.ms=6000

# useradd kafka
# touch /etc/init.d/kafka
# chmod 755 /etc/init.d/kafka

# vi /etc/init.d/kafka 
#!/bin/sh
#
# chkconfig: 345 99 01
# description: Kafka
#
# File : Kafka
#
# Description: Starts and stops the Kafka server
#

source /etc/rc.d/init.d/functions

KAFKA_HOME=/usr/local/kafka_2.11-0.10.0.1
KAFKA_USER=kafka
# See how we were called.
case "$1" in

start)
echo -n "Starting Kafka:"
/sbin/runuser $KAFKA_USER -c "nohup $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties > /var/log/kafka/server.out 2> /var/log/kafka/server.err &"
echo " done."
exit 0
;;

stop)
echo -n "Stopping Kafka: "
/sbin/runuser $KAFKA_USER -c "ps -ef | grep kafka.Kafka | grep -v grep | awk ‘{print \$2}‘ | xargs kill"
echo " done."
exit 0
;;
hardstop)
echo -n "Stopping (hard) Kafka: "
/sbin/runuser $KAFKA_USER -c "ps -ef | grep kafka.Kafka | grep -v grep | awk ‘{print \$2}‘ | xargs kill -9"
echo " done."
exit 0
;;

status)
c_pid=`ps -ef | grep kafka.Kafka | grep -v grep | awk ‘{print $2}‘`
if [ "$c_pid" = "" ] ; then
echo "Stopped"
exit 3
else
echo "Running $c_pid"
exit 0
fi
;;

restart)
stop
start
;;

*)
echo "Usage: cassandra {start|stop|hardstop|status|restart}"
exit 1
;;

esac

# mkdir -p /var/log/kafka
# chown -R kafka.kafka /var/log/kafka
# chown -R kafka.kafka /usr/local/kafka_2.11-0.10.0.1/logs


# 启动服务
# /etc/init.d/kafka start

# 创建topic
# bin/kafka-topics.sh --create --zookeeper 192.168.190.152:2181,192.168.190.153:2181,192.168.190.154:2181 --replication-factor 2 --partitions 1 --topic test333

# 删除
# bin/kafka-topics.sh --delete --zookeeper 192.168.190.152:2181,192.168.190.153:2181,192.168.190.154:2181 --topic test333

# 查看topic
# bin/kafka-topics.sh --list --zookeeper localhost:2181


# 开启生产端
# bin/kafka-console-producer.sh --broker-list 192.168.190.152:9092,192.168.190.153:9092,192.168.190.154:9092 --topic test333


# 开启消费端
# bin/kafka-console-consumer.sh --zookeeper 192.168.190.152:2181,192.168.190.153:2181,192.168.190.154:2181 --topic test333 --from-beginning


# bin/kafka-topics.sh --describe --zookeeper 192.168.190.152:2181,192.168.190.153:2181,192.168.190.154:2181 --topic test333