nohup ./zookeeper-server-start.sh ../config/zookeeper.properties &
或
nohup ./bin/zookeeper-server-start.sh ./config/zookeeper.properties &

nohup(不挂断地运行命令，可以自己百度一下看看，就是nohup 命令 &,让命令在后台执行)

启动kafka

nohup ./kafka-server-start.sh ../config/server.properties &
或
nohup ./bin/kafka-server-start.sh ./config/server.properties &

创建topic

./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
或
./kafka-topics.sh --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test2 --create

列出topic

./kafka-topics.sh --list --zookeeper localhost:2181
或
./kafka-topic --zookeeper localhost:2181 --list

启动生产者并发送消息

./kafka-console-producer.sh --broker-list localhost:9092 --topic zyl
输入消息:
nihao
china

另外开个终端，启动消费者接受消息

./kafka-console-consumer.sh --zookeeper localhost:2181 --topic zyl --from-beginning
可以接收到:
nihao
china

另外开个终端，可以看到有4个进程（分别为zk、kafka、控制台消费者、控制台生产者）

[root[@aa](http://my.oschina.net/test2) bin]# jps | grep -v Jps
9802 QuorumPeerMain
10392 Kafka
10956 ConsoleConsumer
10887 ConsoleProducer

搭建集群（3个节点）

拷贝并修改配置文件

cp server.properties server1.properties
cp server.properties server2.properties
cp server.properties server3.properties

nohup ./kafka-server-start.sh …/config/server1.properties &
nohup ./kafka-server-start.sh …/config/server2.properties &
nohup ./kafka-server-start.sh …/config/server3.properties &

[root@aa bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic（注意，指定的是3）

Created topic “my-replicated-topic”.

[root@aa bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 2,3,0 Isr: 2,3,0

[root@aa bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 4 --partitions 1 --topic my-replicated-topic
Error while executing topic command Topic “my-replicated-topic” already exists.
kafka.common.TopicExistsException: Topic “my-replicated-topic” already exists.
at kafka.admin.AdminUtils$.createOrUpdateTopicPartitionAssignmentPathInZK(AdminUtils.scala:171)atkafka.admin.AdminUtils$.createTopic(AdminUtils.scala:156)
at kafka.admin.TopicCommand$.createTopic(TopicCommand.scala:86)atkafka.admin.TopicCommand$.main(TopicCommand.scala:50)
at kafka.admin.TopicCommand.main(TopicCommand.scala)

[root@aa bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 4 --partitions 1 --topic my-replicated-topic2
Created topic “my-replicated-topic2”.

[root@aa bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic2
Topic:my-replicated-topic2 PartitionCount:1 ReplicationFactor:4 Configs:
Topic: my-replicated-topic2 Partition: 0 Leader: 1 Replicas: 1,2,3,0 Isr: 1,2,3,0

[root@aa bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic zyl
Topic:zyl PartitionCount:1 ReplicationFactor:1 Configs:
Topic: zyl Partition: 0 Leader: 0 Replicas: 0 Isr: 0

启动生产者并发送消息

./kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic

另外开个终端，启动消费者接受消息

./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic

另外开个终端，可以看到有9个进程（分别为1个zk、4个kafka、2个控制台消费者、2个控制台生产者）

  [root@aa ~]# jps | grep -v Jps9802 QuorumPeerMain10392 Kafka11547 Kafka12007 ConsoleProducer10956 ConsoleConsumer10887 ConsoleProducer11469 Kafka12054 ConsoleConsumer11710 Kafka

查看单独的进程：

  ps -ef | grep server.properties | grep -v grepps -ef | grep "server1.properties" | grep -v grep

测试leader

杀掉进程，然后–describe查看

  [root@aaa bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topicTopic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 2,3,0 Isr: 2,3,0可以看到leader是2.

[root@aa bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic2
Topic:my-replicated-topic2 PartitionCount:1 ReplicationFactor:4 Configs:
Topic: my-replicated-topic2 Partition: 0 Leader: 1 Replicas: 1,2,3,0 Isr: 1,2,3,0
可以看到leader是2.

[root@aa bin]# ps -ef | grep server1.properties | grep -v grep

杀掉某个kafka进程

  kill -9 11469

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic2
Topic:my-replicated-topic2 PartitionCount:1 ReplicationFactor:4 Configs:
Topic: my-replicated-topic2 Partition: 0 Leader: 2 Replicas: 1,2,3,0 Isr: 2,3,0

[root@aa bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 2,3,0 Isr: 2,3,0

可以看到my-replicated-topic2的leader变成了2。

问题：
1.zookeeper-shell.sh如何使用
2.能不能通过kafka-server-stop.sh停止某个kafka呢

Command must include exactly one action: --list, --describe, --create or --alter

windows版本

环境搭建

1. jdk环境，上文已经提到过，并配置环境变量，这里不讨论

2. 安装zookeeper，这是运行kafka之前需要的环境(但是可以直接去第三步，因为，kafka有自带的zookeeper服务)
   2.1. 到http://mirror.bit.edu.cn/apache/zookeeper/下载一个版本的zookeeper即可
   2.2. 解压到一个文件夹下，可自己选择
   2.3. 将解压后的文件夹里，conf文件夹下zoo_sample.cfg 改名为 zoo.cfg
   2.4. 修改（如果没有则新增设置)
   dataDir和dataLogDir路径dataDir=D:/Program Files/zookeeper-3.4.13/data/logs
   dataLogDir=D:/Program Files/zookeeper-3.4.13/data/logs
   
   2.5. 设置环境变量：
   新增ZOOKEEPERHOME：D:\Program Files\zookeeper-3.4.13
   Path添加：%ZOOKEEPERHOME%\bin;
   2.6. 检查安装成功与否：

   打开cmd窗口，输入zkserver，结果应该如下，则安装以及配置环境变量成功：
   

3. 安装kafka

   3.1 到http://kafka.apache.org/downloads.html下载

   3.2 解压到自己的文件夹路径（如D:\Program Files\kafka_2.11-1.1.0）

   3.3.1 修改配置文件在config文件夹下，D:\Program Files\kafka_2.11-1.1.0\server.properties

   3.3.2 由于kafka有自带的zookeeper服务，所以如果跳过第2步，可以直接使用自带zk服务，记得配置对应的配置文件/config/zookeeper.properties

   3.4.1 修改server.properties里的log.dirs:
   

   3.4.2 修改zookeeper.properties
   

   3.5 注意观察，bin目录下有一个window文件夹，windows文件夹下的文件内容和bin下的是一样的，只是文件后缀不同，bin目录下的.sh文件需要用shell执行，bat则通过cmd窗口即可执行，当然bat文件通过shell也是可以执行的，至于两者区别，就自行百度了。调出shell窗口的方法：按住shift键+在文件夹空白处单机鼠标右键，就可以看到“在此处打开powershell窗口"的选项。

   3.6 在bin目录下启动powerShell,然后入去一下命令，启动kafka：

   如果是先装了zookeeper，则可以直接执行：
   ./windows/kafka-server-start.bat "D:/Program Files/kafka_2.11-1.1.0/config/ser
   ver.properties"[注意：bat 和 "D:/" 之间有一个空格]
   (这里，由于路径名Programe Files里包含空格，如果不用双引号引用后面的参数，会报错)
   当然，也可以到bin目录的上一层，也就是kafka_2.11-1.1.0目录下执行下面的语句，效果一样：
   ./bin/windows/kafka-server-start.bat ./config/ser
   ver.properties"[注意：bat 和 "./config" 之间有一个空格]
   

如果没有装zookeeper，而是使用自带的，则先执行：
./windows/zookeeper-server-start.bat “D:/Program Files/kafka_2.11-1.1.0/config/zookeeper.properties”

启动自带zookeeper时，可以看到：

然后重新打开一个窗口，开启kafka服务（./windows/kafka-server-start.bat “D:/Program Files/kafka_2.11-1.1.0/config/ser
ver.properties”）时，可以看到：

说明成功了。

• kafka的简单使用：

  4.1 创建一个名字为demo的topic,指定其分区数目为1,副本工厂数目为1：

  到bin/window目录下执行：
  

  ./kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic demo
  输出：Created topic "demo".
  

• 

  4.2 查看所有的topics:

  ./kafka-topics.bat --list --zookeeper localhost:2181
  

  

  如果想删除topics, 可以参考文章：kafka如何彻底删除topic及数据

  4.3 向指定topic发送消息，首先打开2个新的powerShell或cmd窗口，一个生产者producer，一个消费者Consumer：

  设置生产者,设置端口号，并指定topic：
  ./kafka-console-producer.bat --broker-list localhost:9092 --topic demo
  设置消费者，设置从哪一个端口号(2181)、哪个topic(demo)、什么位置(beginning)开始取出消息：
  ./kafka-console-consumer.bat --zookeeper localhost:2181 --topic demo --from-beginning
  

  在producer窗口里输入消息，就可以在对应的Consumer窗口看到对应的消息

  

  好了，大概就这些，应该可以让各位对kafka的概念和基本使用有了一个不错的理解，哈哈