所有操作默认非root用户，本文中用 lz 用户。

本文参考kafka中文社区

kafka自带zookeeper，运行顺序是先zookeeper再kafka：

~$ cd /usr/local/kafka
/usr/local/kafka$ bin/zookeeper-server-start.sh config/zookeeper.properties  # 运行zookeeper

新建终端

~$ cd /usr/local/kafka
/usr/local/kafka$ bin/kafka-server-start.sh config/server.properties  # 运行kafka

假如zookeeper后台没了，所有kafka的broker都会一直报错INFO Socket error occurred如下,直到重新将zookeeper运行起来：

[2019-06-05 23:48:27,418] INFO Socket error occurred: localhost/127.0.0.1:2181: Connection refused (org.apache.zookeeper.ClientCnxn)
[2019-06-05 23:48:29,079] INFO Opening socket connection to server localhost/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2019-06-05 23:48:29,079] INFO Socket error occurred: localhost/127.0.0.1:2181: Connection refused (org.apache.zookeeper.ClientCnxn)
[2019-06-05 23:48:30,641] INFO Opening socket connection to server localhost/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
...

# 重新将zookeeper跑起来

[2019-06-05 23:48:30,641] INFO Socket connection established to localhost/127.0.0.1:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2019-06-05 23:48:30,644] INFO Session establishment complete on server localhost/127.0.0.1:2181, sessionid = 0x1000001acd80014, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
[2019-06-05 23:48:30,644] INFO [ZooKeeperClient] Connected. (kafka.zookeeper.ZooKeeperClient)
[2019-06-05 23:48:30,674] INFO [ReplicaFetcherManager on broker 0] Removed fetcher for partitions Set(my-replicated-topic-0) (kafka.server.ReplicaFetcherManager)
[2019-06-05 23:48:30,718] INFO [ReplicaFetcher replicaId=0, leaderId=1, fetcherId=0] Starting (kafka.server.ReplicaFetcherThread)
[2019-06-05 23:48:30,719] INFO [ReplicaFetcherManager on broker 0] Added fetcher to broker BrokerEndPoint(id=1, host=ubuntu:9093) for partitions Map(my-replicated-topic-0 -> (offset=2, leaderEpoch=6)) (kafka.server.ReplicaFetcherManager)
[2019-06-05 23:48:30,735] INFO [Log partition=my-replicated-topic-0, dir=/tmp/kafka-logs] Truncating to 2 has no effect as the largest offset in the log is 1 (kafka.log.Log)