1. 启动Zookeeper服务器
./zookeeper-server-start.sh /opt/cx/kafka_2.11-0.9.0.1/config/zookeeper.properties
2. 修改broker-1,broker-2配置
broker.id=1 listeners=PLAINTEXT://:9093 # The port the socket server listens on port=9093 log.dirs=/opt/cx/kafka/kafka-logs-1
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服务器
./kafka-server-start.sh /opt/cx/kafka/config/server-1.properties &
./kafka-server-start.sh /opt/cx/kafka/config/server-2.properties &
4. 查看broker集群
ps -ef|grep kafka
5. 创建并查看topic
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 1 --topic my-replicated-topic Created topic "my-replicated-topic".
查看topic
./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),因为副本只有在集群服务器下才会创建。
./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. 发送消息
./kafka-console-producer.sh --broker-list localhost:9093 --topic my-replicated-topic My replicated message 1 My replicated message 2
7. 接收消息
./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已经不在同步的复制集中。
[root@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 [root@Server bin]# kill -9 11351 [root@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写入的。
[root@Server bin]# ./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic My replicated message 1 My replicated message 2
本文出自 “这个人的IT世界” 博客,请务必保留此出处http://favccxx.blog.51cto.com/2890523/1763046
原文地址:http://favccxx.blog.51cto.com/2890523/1763046