kafka operation

运行环境：mac os

1. 启动zookeeper

./bin/zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties

2. 启动kafka服务

./bin/kafka-server-start /usr/local/etc/kafka/server.properties

3. 查看topic列表

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

4. 创建topic

./bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic hezhixiong

--create：指定创建topic动作
--zookeeper：指定kafka连接zk的地址
--replication-factor：指定每个分区的复制因子个数，默认1个
--partitions：指定当前创建的kafka分区数量，默认为1个
--topic：指定新建topic的名称

5. 查看topic的描述信息

./bin/kafka-topics --describe --zookeeper localhost:2181 --topic hezhixiong

6. 修改topic信息

./bin/kafka-topics --zookeeper localhost:2181 --alter --topic hezhixiong --partitions 4
<!-- Kafka分区数量只允许增加，不允许减少 -->

7. 删除topic

./bin/kafka-topics --zookeeper localhost:2181 --delete --topic hezhixiong
Topic hezhixiong is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

默认情况下Kafka的Topic是没法直接删除的，而是标记删除而已。如果需要彻底删除topic，有以下两种方式：
1. 通过delete命令删除后，手动将本地磁盘以及zk上的相关topic的信息删除
2. 配置server.properties文件，给定参数delete.topic.enable=true，重启kafka服务，此时执行delete命令表示允许进行Topic的删除

8. 生产者

./bin/kafka-console-producer --broker-list localhost:9092 --topic hezhixiong

9. 消费者

./bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic hezhixiong --from-beginning

 10. broker集群

　　首先为每个broker创建一个配置文件：

cd /usr/local/etc/kafka
cp server.properties server_1.properties
cp server.properties server_2.properties

vi server_1.properties
    broker.id=1 
    listeners=PLAINTEXT://:9093 
    log.dir=/tmp/kafka-logs/one
vi server_2.properties
    broker.id=2 
    listeners=PLAINTEXT://:9094 
    log.dir=/tmp/kafka-logs/two

　　broker.id是集群中每个节点的唯一且永久的名称，我们修改端口和日志目录是因为我们现在在同一台机器上运行，我们要防止broker在同一端口上注册和覆盖对方的数据。

　　我们已经运行了zookeeper和刚才的一个kafka节点，所有我们只需要在启动2个新的kafka节点。

./bin/kafka-server-start /usr/local/etc/kafka/server_1.properties
./bin/kafka-server-start /usr/local/etc/kafka/server_2.properties

　　现在，我们创建一个新topic，把备份设置为3

./bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic hzx-new
./bin/kafka-topics --describe --zookeeper localhost:2181 --topic hzx-new  查看topic得到输出如下：
Topic:hzx-new PartitionCount:1 ReplicationFactor:3 Configs:
       Topic: hzx-new Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0

从上面的输出，我们可以得出结论，第一行给出所有分区的摘要，显示主题名称，分区数量和我们已经选择的复制因子。在第二行中，每个节点将是分区的随机选择部分的领导者。

在我的例子中，我看到的第一个broker(with broker.id 2)是领导者。 然后Replicas:2，1，0 意味着所有代理复制主题。最后 Isr 是 in-sync 副本的集合。 那么，这是副本的子集，当前活着并被领导者赶上。

"leader"：该节点负责该分区的所有的读和写，每个节点的leader都是随机选择的。
"replicas"：备份的节点列表，无论该节点是否是leader或者目前是否还活着，只是显示。
"isr"：“同步备份”的节点列表，也就是活着的节点并且正在同步leader。

　　启动生产者和消费者，在生产者端发送消息，在消费者端能看到消息的消息

./bin/kafka-console-producer --broker-list localhost:9092 --topic hzx-new   （生产者）
./bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic hzx-new --from-beginning  （消费者）

 　　体验并测试kafka集群的容错，目前Leader是broker2，所以结束调broker2，并查看topic：hzx-new 的信息，输出如下

Topic:hzx-new    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: hzx-new    Partition: 0    Leader: 1    Replicas: 2,1,0    Isr: 1,0

可以看出 broker2已经不再同步备份集合里了，备份节点之一的broker1成为了新的leader了。

参考资料：

http://orchome.com/6