【Kafka入门】搭建Kafka本地环境

时间:2024-03-09 07:35:38

本博文介绍如何一步步搭建起Kafka本地环境。

下载Kafka 0.9.0.0 并配置软链接

  下载好后,放入电脑本地安装目录,mac下我放在/usr/local下,解压Kafka。

tar -xzf kafka_2.11-0.9.0.0.tgz

  然后建立一个当前版本的kafka的软链接,这样在配置的时候系统环境配置的时候不需要写版本号,也有利于以后的版本更新。

ln -s kafka_2.11-0.9.0.0 kafka

  接着cd kafka进入主目录。

 

启动Kafka Server

  在启动前,需要安装并配置Zookeeper,不过在kafka中带有一个单节点示例的Zookeeper实例脚本包,所以我们也可以先不安装配置Zookeeper,在启动Kafka Server前,我们先启动Zookeeper:

bin/zookeeper-server-start.sh config/zookeeper.properties

  在Zookeeper成功启动以后,就可以启动Kafka Server了:

bin/kafka-server-start.sh config/server.properties

 

创建topic话题

  通过下面的脚本,我们创建了一个名为test的topic话题,该话题分区数为1,冗余份数为1。然后通过list展示当前话题,会列出test。  

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

 

启动生产者Producer和消费者Consumer

  通过以下脚本启动一个Producer,然后可以输入生产的消息:

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

this is a test of producer

   通过下面的命令启动消费者Consumer来消费,然后会打印出上述输入的消息:

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
this is a test of producer

  如果上述两个命令,在不同的终端启动,这时候你在生产者终端输入消息并回车,消费者端可以看到。

 

启动多broker的集群

  由于Kafka是分布式的消息系统,目前我们只是启动了一个Kafka Server,并看不到分布式的效果,下面在本地启动多个Kafka Server,首先我们需要将server的配置文件复制两份,因为每个Kafka服务启动都是依赖配置文件,在配置文件中设置了诸如服务启动端口等信息。

cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties

  修改上述复制出来的配置文件,由于默认服务启动在9092端口,所以后续两个server启动在9093和9094;然后默认的server的broker id为0,由于id是borker的唯一标识,所以后续的id依次为1和2.默认的几个属性如下:

broker.id=0

############################# Socket Server Settings #############################

listeners=PLAINTEXT://:9092

# The port the socket server listens on
#port=9092

# A comma seperated list of directories under which to store log files
log.dirs=/tmp/kafka-logs

  server1和server2的关于这几个属性的配置修改如下:

broker.id=1
listeners=PLAINTEXT://:9093
port=9093
log.dirs=/tmp/kafka-logs-1

broker.id=2
listeners=PLAINTEXT://:9094
port=9094
log.dirs=/tmp/kafka-logs-2

  然后依次启动两个broker的服务:

bin/kafka-server-start.sh config/server-1.properties 
bin/kafka-server-start.sh config/server-2.properties

  最好在多个终端启动,分别看看效果。

  接着我们启动一个冗余数为3的topic并describe这个topic,效果显示如下:

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replication-topic
Created topic "my-replication-topic".
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replication-topic
Topic:my-replication-topic    PartitionCount:1    ReplicationFactor:3    Configs:
Topic: my-replication-topic    Partition: 0    Leader: 0    Replicas: 0,2,1    Isr: 0,2,1

  显示的内容一共是两行,第一行是对所有partition的总结,后续各行是每个partition的信息,由于我们这里只有一个partition,所以后续只有1行(也就是第二行)

  • Leader:表示负责该partition所有读写的节点
  • Replicas:所有拥有该partition 日志的备份数据的节点,不管这些节点是否是Leader或者活着。
  • Isr:是所有同步Replicas集合。 这是Replicas列表的子集合,在这个集合中的都是活着并且赶上Leader进度的。

多broker的容错性测试

  上面的my-replication-topic的Leader是0号服务,那我们尝试找到0号broker并将该进程杀死,脚本如下:

Lilis-MacBook-Pro:kafka_2.10-0.9.0.0 lili$ ps | grep server.properties
 2366 ttys001    1:28.51 /Library/Java/JavaVirtualMachines。。。
 2686 ttys006    0:00.00 grep server.properties
Lilis-MacBook-Pro:kafka_2.10-0.9.0.0 lili$ kill -9 2366

  再describe my-replication-topic的时候,发现leader已经变了,此时变成了2号broker,而且同步Replicas集合中也少了0号broker。

Lilis-MacBook-Pro:kafka_2.10-0.9.0.0 lili$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replication-topic
Topic:my-replication-topic    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: my-replication-topic    Partition: 0    Leader: 2    Replicas: 0,2,1    Isr: 2,1
Lilis-MacBook-Pro:kafka_2.10-0.9.0.0 lili$

  当然,再次启动消费者来消费消息,消息依然还是存在的。

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