3.5.2 kafka分布式消息队列-Kafka集群环境搭建和命令行操作

痛定思痛。 2022-12-28 11:25 180阅读 0赞

**目录**

1.搭建Kafka集群环境

1.1 下载

1.2 解压

1.3 配置环境变量

1.4 修改配置

1.5 分发安装包

1.6 启动

1.7 验证

2.群起脚本

3.kafka命令行操作

1）查看当前服务器中的所有 topic

2）创建 topic

3）删除 topic

4）发送消息

5）消费消息

6）查看某个 Topic 的详情

7）修改分区数

--------------------

# 1.搭建Kafka集群环境 #

## 1.1 下载 ##

下载对应版本 Kafka，这里我下载的最新稳定版本 `2.4.1`。官方下载地址：http://kafka.apache.org/downloads

> 这里解释一下 kafka 安装包的命名规则：以 `kafka_2.11-2.4.1.tgz` 为例，前面的 2.11 代表 Scala 的版本号（Kafka 采用 Scala 语言进行开发），后面的 2.4.1 则代表 Kafka 的版本号。

## 1.2 解压 ##

[xiaokang@hadoop01 ~]$ tar -zxvf kafka_2.11-2.4.1.tgz -C /opt/software/
    kafka_2.11-2.4.1/
    
    [xiaokang@hadoop01 software]$ mv /opt/software/kafka_2.11-2.4.1/ /opt/software/kafka-2.4.1

## 1.3 配置环境变量 ##

编辑 `profile` 文件：

[xiaokang@hadoop01 ~]$ sudo vim /etc/profile

在原来基础上更新配置环境变量：

export KAFKA_HOME=/opt/software/kafka-2.4.1
    export PATH=${JAVA_HOME}/bin:${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:${ZOOKEEPER_HOME}/bin:${HIVE_HOME}/bin:${HBASE_HOME}/bin:${KYLIN_HOME}/bin:${KAFKA_HOME}/bin:$PATH

使得配置的环境变量生效：

[xiaokang@hadoop01 ~]$ source /etc/profile

## 1.4 修改配置 ##

在 Kafka 安装目录下创建`kafka-logs`文件夹（用来存储分区信息的，不要把它与存放错误日志的目录混淆了，日志目录是配置在 `log4j.properties` 文件 里的）

[xiaokang@hadoop01 ~]$ mkdir /opt/software/kafka-2.4.1/kafka-logs

进入安装目录的 `config/` 目录下，修改配置文件`server.properties`：

# broker的全局唯一标识号，不能重复. 给集群中的每个broker配置一个不同的id
    broker.id=0
    # 分区数据的存储位置
    log.dirs=/opt/software/kafka-2.4.1/kafka-logs
    # 连接Zookeeper集群地址
    zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181

## 1.5 分发安装包 ##

分发之后需要修改 **broker.id( id 值一定不能重复 )**的值，同时建议其他节点的环境变量也配置下。

[xiaokang@hadoop01 ~]$ scp -r /opt/software/kafka-2.4.1/ xiaokang@hadoop02:/opt/software
    [xiaokang@hadoop01 ~]$ scp -r /opt/software/kafka-2.4.1/ xiaokang@hadoop03:/opt/software

## 1.6 启动 ##

使用下面命令启动即可：

#先启动3个节点的zookeeper
    zkServer.sh start
    #再启动3个节点的kafka
    kafka-server-start.sh /opt/software/kafka-2.4.1/config/server.properties

关闭kafka集群

[atguigu@hadoop102 kafka]$ kafka-server-stop.sh stop
    [atguigu@hadoop103 kafka]$ kafka-server-stop.sh stop
    [atguigu@hadoop104 kafka]$ kafka-server-stop.sh stop

## 1.7 验证 ##

创建测试主题：

[xiaokang@hadoop01 ~]$ kafka-topics.sh --create --zookeeper hadoop01:2181 --replication-factor 3 --partitions 1 --topic xiaokang
    
    #kafka-topics.sh 任何和 topic 相关的操作都使用这个命令
    #--create 表示创建一个 topic
    #--zookeeper 指明任意一个 zookeeper 服务器地址
    #--replication-factor 表示每个 topic 的副本数. 注意: 副本数必须小于等于 kafka 集群的数量.
    #--partitions 这个 topic 的分区的数量
    #--topic 这个 topic 的名字.

创建完成后可以使用以下命令查看创建的主题信息：

[xiaokang@hadoop01 ~]$ kafka-topics.sh --describe --zookeeper hadoop01:2181 --topic xiaokang

![图片][aaef94e8ad87eacd9a77c9a5e19ae88a.png]

可以看到分区 0 的有 0,1,2 三个副本，且三个副本都是可用副本，都在 ISR(in-sync Replica 同步副本) 列表中，其中 2 为首领副本，此时代表集群已经搭建成功。

# 2.群起群关脚本 #

#! /bin/bash
    case $1 in
    "start"){
        for i in h2node01 h2node02 h2node03
        do
            echo " --------启动 $i Kafka-------"
            ssh $i "kafka-server-start.sh -daemon /opt/software/kafka-2.4.1/config/server.properties"
        done
    };;
    "stop"){
        for i in h2node01 h2node02 h2node03
        do
            echo " --------停止 $i Kafka-------"
            ssh $i "/kafka-server-stop.sh stop"
        done
    };;
    esac

# 3.kafka命令行操作 #

## 1）查看当前服务器中的所有 topic ##

bin/kafka-topics.sh --zookeeper hadoop102:2181 --list

## 2）创建 topic ##

副本数不能超过集群中机器的个数

bin/kafka-topics.sh --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 1 --topic first

选项说明：

*  \--topic 定义 topic 名
 *  \--replication-factor 定义副本数
 *  \--partitions 定义分区数

## 3）删除 topic ##

bin/kafka-topics.sh --zookeeper hadoop102:2181 --delete --topic first

需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除。

## 4）发送消息 ##

[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --brokerlist hadoop102:9092 --topic first
    >hello world
    >atguigu atguigu

## 5）消费消息 ##

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
    --zookeeper hadoop102:2181 --topic first
    [atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
    --bootstrap-server hadoop102:9092 --topic first
    [atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
    --bootstrap-server hadoop102:9092 --from-beginning --topic first

\--from-beginning：会把主题中以往所有的数据都读取出来。

## 6）查看某个 Topic 的详情 ##

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper
    hadoop102:2181 --describe --topic first

## 7）修改分区数 ##

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper
    hadoop102:2181 --alter --topic first --partitions 6

[aaef94e8ad87eacd9a77c9a5e19ae88a.png]: /images/20221120/fd51d2e44724422491caf60b7cd7af71.png