搭建Kafka集群

213123

1. 启动Zookeeper服务器

1	./zookeeper-server-start.sh /opt/cx/kafka_2.11-0.9.0.1/config/zookeeper.properties

2. 修改broker-1，broker-2配置

1 2 3 4 5

broker.id=1 listeners=PLAINTEXT://:9093 # The port the socket server listens on port=9093 log.dirs=/opt/cx/kafka/kafka-logs-1

1 2 3 4 5

broker.id=2 listeners=PLAINTEXT://:9094 # The port the socket server listens on port=9094 log.dirs=/opt/cx/kafka/kafka-logs-2

3. 启动broker-1，broker-2服务器

1	./kafka-server-start.sh /opt/cx/kafka/config/server-1.properties &

1	./kafka-server-start.sh /opt/cx/kafka/config/server-2.properties &

4. 查看broker集群

1	ps -ef|grep kafka

5. 创建并查看topic

1 2	./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 1 --topic my-replicated-topic Created topic "my-replicated-topic".

查看topic

1 2 3

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

从控制台可以看出，总共创建了一个主题（第一行是汇总行，下面列表显示的是详细信息）。
    leader：从分区中读写消息的节点。每个节点成为leader都是随机的。
    replicas：复制日志的节点，不管它是否是leader，也不管它是否还可用。
    isr：同步状态的副本集。这个副本集包括活动者的以后可能成为leader的节点。
    现在看一下在单节点创建的主题test（如果没有更好Zookeeper的话，之前创建的主题都存在）。我们发现根本没有副本（Replicas: 0），因为副本只有在集群服务器下才会创建。

1 2 3

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test Topic:test    PartitionCount:1    ReplicationFactor:1    Configs:     Topic: test    Partition: 0    Leader: 0    Replicas: 0    Isr: 0

6. 发送消息

1 2 3

./kafka-console-producer.sh --broker-list localhost:9093 --topic my-replicated-topic My replicated message 1 My replicated message 2

7. 接收消息

1 2 3

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

8. 测试容错机制，结束server-1的进程，查看集群主题。可以看到Leader节点已经转移，而且server-1已经不在同步的复制集中。

1 2 3 4 5 6 7

[iyunv@Server bin]# ps -ef|grep server-1.properties root      5777 14396  0 12:07 pts/5    00:00:00 grep server-1.properties root     11351 10096  1 11:08 pts/3    00:01:01 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC -Djava.awt.headless=true -Xloggc:/opt/cx/kafka_2.11-0.9.0.1/bin/../logs/kafkaServer-gc.log -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dkafka.logs.dir=/opt/cx/kafka_2.11-0.9.0.1/bin/../logs -Dlog4j.configuration=file:./../config/log4j.properties -cp :/opt/cx/kafka_2.11-0.9.0.1/bin/../libs/* kafka.Kafka /opt/cx/kafka/config/server-1.properties [iyunv@Server bin]# kill -9 11351 [iyunv@Server bin]# ./kafka-topics.sh  --describe --zookeeper localhost:2181 --topic my-replicated-topic Topic:my-replicated-topic    PartitionCount:1    ReplicationFactor:2    Configs:     Topic: my-replicated-topic    Partition: 0    Leader: 2    Replicas: 1,2    Isr: 2

9. 重新运行消费者仍能接收消息，尽管这些消息是由leader-1写入的。

1 2 3

[iyunv@Server bin]# ./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic My replicated message 1 My replicated message 2