【Linux】ubuntu下kafka集群环境搭建

时间:2022-05-18 02:06:30

1.下载kafka正式版,本文使用kafka_2.10-0.8.2.2(在此特别说明下,kafka_2.9.2-0.8.1.1.tgz由于版本相对较低的问题,好多.sh文件时缺少的,比如kafka-topics.sh,所以建议大家在安装的时候不要使用太老的版本,问题较多,并且启动kafka-server-start的时候还报错,始终无法启动成功~)

官网下载链接如下:http://kafka.apache.org/downloads.html

2.解压tar包

命令如下:tar -zxvf kafka_2.10-0.8.2.2.tar.gz

3.修改server.properties相关配置

进入kafka根目录下的config/server.properties,主要修改的3个参数如下:

broker.id=0 //broker的标识(正数),集群内各个broker.id不能重复。

port=9092 //端口号,单节点内各个port不能重复(为了方便看最好不同节点端口也不要重复)

log.dir =/tmp //存储log的地方(数据文件)。可根据自己的日志目录进行修改

以上配置时在单机上的配置,多台机器的配置要根据ip进行相关配置;


4.修改config下面的log4j.properties、zookeeper.properties的日志目录;

    log4j.properties下面的kafka.logs.dir=logs,可根据自己需要修改成自己的目录;

    zookeeper.properties下面的dataDir=/tmp/zookeeper,可根据需要修改成自己的目录

5.验证是否安装成功

(1)  start server

            启动一个单节点的zookeeper,命令如下:>bin/zookeeper-server-start.sh config/zookeeper.properties

    准备启动一个3个broker节点的kafka集群,因此做如下配置:

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

   并做如下修改:

   config/server-1.properties:
        broker.id=1
        port=9093
        log.dir=/tmp/kafka-logs-1

   config/server-2.properties:
        broker.id=2
        port=9094
        log.dir=/tmp/kafka-logs-2

   说明:
   broker.id: broker节点的唯一标识
   port: broker节点使用端口号
   log.dir: 消息目录位置

   启动3个broker节点
   $ JMX_PORT=9997 bin/kafka-server-start.sh config/server-1.properties &
   $ JMX_PORT=9998 bin/kafka-server-start.sh config/server-2.properties &
   $ JMX_PORT=9999 bin/kafka-server-start.sh config/server.properties &

   标红的内容可根据自己的需要进行配置;

  启动单个broker节点的kafka集群,命令如下:>bin/kafka-server-start.sh config/server.propertie

      第一行时启动kafka自带的zookeeper

(2)create a topic

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

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

topic: 3test    partition: 0    leader: 2       replicas: 2,1,0 isr: 2,1,0
 topic: test     partition: 0    leader: 0       replicas: 0     isr: 0
 topic: test_topic       partition: 0    leader: 1       replicas: 0,1,2 isr: 1,2,0
 说明:
 partiton: partion id,由于此处只有一个partition,因此partition id 为0
 leader:当前负责读写的lead broker id
 relicas:当前partition的所有replication broker  list
 isr:relicas的子集,只包含出于活动状态的broker

(3)send message

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

(4)start a consumer

$bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

kafka-console-producer.shkafka-console-cousumer.sh只是系统提供的命令行工具。这里启动是为了测试是否能正常生产消费,验证流程正确性,在实际开发中还是要自行开发自己的生产者与消费者。

(5)start a produce


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

   这样就可以的命令行输入信息,可以看到consumer可以获得同样的信息~


  6.干掉follow broker

      杀掉一个非leader broker命令如下:server-1.properties里面的broker =1

      $ pkill -9 -f server-1.properties

    查看topic:
   $ bin/kafka-list-topic.sh --zookeeper localhost:2181
   topic: 3test    partition: 0    leader: 2       replicas: 2,1,0 isr: 2,0
   topic: test     partition: 0    leader: 2       replicas: 0,1,2 isr: 2,0
   此时,存活的broker只有2,0
   测试:produce发送消息,consumer能正常接收到

 7.继续干掉leader broker

   干掉leader broker后,连续查看topic状态
   $ pkill -9 -f server-2.properties                 
   $ bin/kafka-list-topic.sh --zookeeper localhost:2181
   topic: test    partition: 0    leader: 2       replicas: 2,1,0 isr: 2,0
 
   $ bin/kafka-list-topic.sh --zookeeper localhost:2181
   topic: test    partition: 0    leader: 2       replicas: 2,1,0 isr: 2,0
 
   $ bin/kafka-list-topic.sh --zookeeper localhost:2181
   topic: test    partition: 0    leader: 0       replicas: 2,1,0 isr: 0
 
   $ bin/kafka-list-topic.sh --zookeeper localhost:2181
   topic: test    partition: 0    leader: 0       replicas: 2,1,0 isr: 0
 
   杀掉leader broker过了一会,broker 0成为新的leader broker
   测试:produce发送消息,consumer能正常接收到

8.kafka常用命令

创建topic

/kafka-topics.sh --create --zookeeper 192.168.153.128:2181 --replication-factor 1--partitions 1 --topic test123


查看topic信息

./kafka-topics.sh --describe --topic test123 --zookeeper 192.168.153.128:2181


修改topic分区

./kafka-topics.sh --alter --topic test123 --partitions 2 --zookeeper 192.168.153.128:2181


删除topic

./kafka-run-class.shkafka.admin.DeleteTopicCommand --topic test1 --zookeeper192.168.35.122:2181,192.168.35.123:2181

只是删除了zookeeper中元数据,数据文件还需手动删除。