聊聊 Kafka：在 Linux 环境上搭建 Kafka

悠悠 2022-09-03 01:20 179阅读 0赞

## 一、环境准备 ##

jdk下载地址链接：[jdk 1.8][]，提取码: dv5h  
zookeeper下载地址链接：[zookeeper3.4.14 ][zookeeper3.4.14]，提取码: 3dch  
kafka下载地址链接：[kafka2.12][]，提取码: 61bc

**1.1 Java环境为前提**

**1.1.1 上传`jdk-8u261-linux-x64.rpm`到服务器并安装**

# 安装命令
    rpm -ivh jdk-8u261-linux-x64.rpm

![在这里插入图片描述][20210211001235318.png]

**1.1.2 配置环境变量**

# 编辑配置文件,jdk的bin目录到/etc/profile文件，对所有用户的shell有效
    vim /etc/profile
    # 生效
    source /etc/profile

![在这里插入图片描述][20210211001914469.png]

export JAVA_HOME=/usr/java/jdk1.8.0_261-amd64
    export PATH=$PATH:$JAVA_HOME/bin

![在这里插入图片描述][2021021100214416.png]

# 验证
    java -version

![在这里插入图片描述][20210211002254431.png]  
至此，jdk安装成功。

**1.2 Zookeeper的安装配置**

**1.2.1 上传`zookeeper-3.4.14.tar.gz`到服务器，解压到`/opt`**

# 解压zk到指定目录
    tar -zxf zookeeper-3.4.14.tar.gz -C /opt

![在这里插入图片描述][20210211003411823.png]  
**1.2.2 修改`Zookeeper`保存数据的目录，`dataDir`**

# 进入conf配置目录
    cd /opt/zookeeper-3.4.14/conf
    # 复制zoo_sample.cfg命名为zoo.cfg
    cp zoo_sample.cfg zoo.cfg
    # 编辑zoo.cfg文件
    vim zoo.cfg
    dataDir=/var/riemann/zookeeper/data

![在这里插入图片描述][20210211004024193.png]  
![在这里插入图片描述][20210211004150924.png]  
**1.2.3 编辑`/etc/profile`，使配置生效**

> 设置环境变量`ZOO_LOG_DIR`，指定`Zookeeper`保存日志的位置；  
> `ZOOKEEPER_PREFIX`指向`Zookeeper`的解压目录；  
> 将Zookeeper的`bin`目录添加到`PATH`中：

![在这里插入图片描述][20210211010033961.png]

export ZOOKEEPER_PREFIX=/opt/zookeeper-3.4.14
    export PATH=$PATH:$ZOOKEEPER_PREFIX/bin
    export ZOO_LOG_DIR=/var/riemann/zookeeper/log

配置完以后再生效配置：

source /etc/profile

**1.2.4 启动`Zookeeper`，确认`Zookeeper`的状态**

zkServer.sh start

![在这里插入图片描述][20210211010250841.png]

至此，zookeeper安装成功。

**1.3 Kafka 的安装与配置**

**1.3.1 上传`kafka_2.12-1.0.2.tgz`到服务器并解压**

tar -zxf kafka_2.12-1.0.2.tgz -C /opt

![在这里插入图片描述][20210211010637345.png]  
**1.3.2 配置环境变量并生效**

vim /etc/profile

export KAFKA=/opt/kafka_2.12-1.0.2
    export PATH=$PATH:$KAFKA/bin

source /etc/profile

**1.3.3 配置`/opt/kafka_2.12-1.0.2/config`中的`server.properties`文件**

vi  /opt/kafka_2.12-1.0.2/config/server.properties

![在这里插入图片描述][20210211012546286.png]

> `Kafka`连接`Zookeeper`的地址，此处使用本地启动的`Zookeeper`实例  
> 连接地址是`localhost:2181`  
> 后面的 `myKafka` 是`Kafka`在Zookeeper中的根节点路径

![在这里插入图片描述][20210211011938942.png]  
配置kafka存储持久化数据目录  
![在这里插入图片描述][20210211012334771.png]

log.dirs=/var/riemann/kafka/kafka-logs

创建上述持久化数据目录

mkdir -p /var/riemann/kafka/kafka-logs

**1.4 启动Kafka**

进入Kafka安装的根目录，执行如下命令：

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

![在这里插入图片描述][20210211013506891.png]  
启动成功，可以看到控制台输出的最后一行的started状态：此时kafka安装成功。

![在这里插入图片描述][20210211013627472.png]

**1.5 重新开一个窗口，查看Zookeeper的节点**  
![在这里插入图片描述][20210211013854618.png]

**1.6 此时Kafka是前台模式启动，要停止，使用Ctrl+C**

如果要后台启动，使用命令：

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

查看Kafka的后台进程：

ps -ef | grep kafka

![在这里插入图片描述][20210211014331157.png]

停止后台运行的Kafka：

kafka-server-stop.sh

## 二、生产与消费 ##

查看zookeeper状态，zookeeper启动成功，再启动kafka。

![在这里插入图片描述][20210211015031215.png]  
**2.1 kafka-topics.sh 用于管理主题**

查看命令的帮助信息

[root@master1 bin]# kafka-topics.sh
    Create, delete, describe, or change a topic.
    Option                                   Description                            
    ------                                   -----------                            
    --alter                                  Alter the number of partitions,        
                                               replica assignment, and/or           
                                               configuration for the topic.         
    --config <String: name=value>            A topic configuration override for the 
                                               topic being created or altered.The   
                                               following is a list of valid         
                                               configurations:                      
                                             	cleanup.policy                        
                                             	compression.type                      
                                             	delete.retention.ms                   
                                             	file.delete.delay.ms                  
                                             	flush.messages                        
                                             	flush.ms                              
                                             	follower.replication.throttled.       
                                               replicas                             
                                             	index.interval.bytes                  
                                             	leader.replication.throttled.replicas 
                                             	max.message.bytes                     
                                             	message.format.version                
                                             	message.timestamp.difference.max.ms   
                                             	message.timestamp.type                
                                             	min.cleanable.dirty.ratio             
                                             	min.compaction.lag.ms                 
                                             	min.insync.replicas                   
                                             	preallocate                           
                                             	retention.bytes                       
                                             	retention.ms                          
                                             	segment.bytes                         
                                             	segment.index.bytes                   
                                             	segment.jitter.ms                     
                                             	segment.ms                            
                                             	unclean.leader.election.enable        
                                             See the Kafka documentation for full   
                                               details on the topic configs.        
    --create                                 Create a new topic.                    
    --delete                                 Delete a topic                         
    --delete-config <String: name>           A topic configuration override to be   
                                               removed for an existing topic (see   
                                               the list of configurations under the 
                                               --config option).                    
    --describe                               List details for the given topics.     
    --disable-rack-aware                     Disable rack aware replica assignment  
    --force                                  Suppress console prompts               
    --help                                   Print usage information.               
    --if-exists                              if set when altering or deleting       
                                               topics, the action will only execute 
                                               if the topic exists                  
    --if-not-exists                          if set when creating topics, the       
                                               action will only execute if the      
                                               topic does not already exist         
    --list                                   List all available topics.             
    --partitions <Integer: # of partitions> The number of partitions for the topic 
                                               being created or altered (WARNING:   
                                               If partitions are increased for a    
                                               topic that has a key, the partition  
                                               logic or ordering of the messages    
                                               will be affected                     
    --replica-assignment <String:            A list of manual partition-to-broker   
      broker_id_for_part1_replica1 :           assignments for the topic being      
      broker_id_for_part1_replica2 ,           created or altered.                  
      broker_id_for_part2_replica1 :                                                
      broker_id_for_part2_replica2 , ...>                                           
    --replication-factor <Integer:           The replication factor for each        
      replication factor>                      partition in the topic being created.
    --topic <String: topic>                  The topic to be create, alter or       
                                               describe. Can also accept a regular  
                                               expression except for --create option
    --topics-with-overrides                  if set when describing topics, only    
                                               show topics that have overridden     
                                               configs                              
    --unavailable-partitions                 if set when describing topics, only    
                                               show partitions whose leader is not  
                                               available                            
    --under-replicated-partitions            if set when describing topics, only    
                                               show under replicated partitions     
    --zookeeper <String: urls>               REQUIRED: The connection string for    
                                               the zookeeper connection in the form 
                                               host:port. Multiple URLS can be      
                                               given to allow fail-over.            
    [root@master1 bin]#

# 列出现有的主题
    [root@master1 ~]# kafka-topics.sh --list --zookeeper localhost:2181/myKafka
    # 创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本。
    [root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_test --partitions 1 --replication-factor 1
    # 查看分区信息
    [root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --list
    # 查看指定主题的详细信息
    [root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_test 
    # 删除指定主题
    [root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_test

列出现有主题，创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本。

![在这里插入图片描述][20210211015707765.png]

查看指定主题的详细信息  
![在这里插入图片描述][20210211020143946.png]  
创建主题，该主题包含多个分区

> 多个分区：横向扩展  
> 多个副本：高可用

![在这里插入图片描述][20210211020847745.png]

**2.2 kafka-console-consumer.sh用于消费消息**

# 开启消费者
    [root@node1 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_test 
    # 开启消费者方式二，从头消费，不按照偏移量消费
    [root@node1 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_test --from-beginning

**2.3 kafka-console-producer.sh用于生产消息**

# 开启生产者
    [root@node1 ~]# kafka-console-producer.sh --topic topic_test --broker-list localhost:9020

**2.4 具体操作**

开启消费者和生产者，生产并消费消息。

![在这里插入图片描述][2021021102181993.png]  
消费者，按照偏移量消费

![在这里插入图片描述][20210211022059441.png]  
消费者从头消费，不按照偏移量消费

![在这里插入图片描述][20210211022324362.png]

[jdk 1.8]: https://pan.baidu.com/s/1r7Ivzk0226wgpN31yxLvHw
[zookeeper3.4.14]: https://pan.baidu.com/s/1sEedYB86VRbyB7jvCPPvBA
[kafka2.12]: https://pan.baidu.com/s/13lHCe5TDxqS3UGCEFTc37w
[20210211001235318.png]: /images/20220829/df9013ae1d3e4116b5f9795089edbca8.png
[20210211001914469.png]: /images/20220829/9d42055ce140494eba0d684cf72acaf9.png
[2021021100214416.png]: /images/20220829/5f905489169548d0ac5164cdff70c690.png
[20210211002254431.png]: /images/20220829/a3046d92be16459b874a5a1acf331df3.png
[20210211003411823.png]: /images/20220829/565a75ea7f9347e8885a09f896265d34.png
[20210211004024193.png]: /images/20220829/49ee6e2533eb4593b113c8eb8d3028f5.png
[20210211004150924.png]: /images/20220829/a616eee79c13423a88df61bd4f7a1961.png
[20210211010033961.png]: /images/20220829/8a8fdd612fc64d63aae6e49655dbc5bc.png
[20210211010250841.png]: /images/20220829/cf8d7f3bc2a9425ab338e5f2fa18dcc3.png
[20210211010637345.png]: /images/20220829/5f305fc712254f6e9ba4c0c8d55aa01b.png
[20210211012546286.png]: /images/20220829/317297d798a44d239f560d5c818ba651.png
[20210211011938942.png]: /images/20220829/39ab43b5f0434ccfbb98957b79a6e826.png
[20210211012334771.png]: /images/20220829/7a5358bd3300473d922505dd3d1d78e1.png
[20210211013506891.png]: /images/20220829/335f2fe93ca246caa7a77847fd9e154d.png
[20210211013627472.png]: /images/20220829/541d1b63addf4a6d86ae96f7efbbbb59.png
[20210211013854618.png]: /images/20220829/f382924082904387a1e0f96d3224d833.png
[20210211014331157.png]: /images/20220829/45b24a22e340425ab2c97121b7007820.png
[20210211015031215.png]: /images/20220829/9a795fa8ad324cafa133d024bc6d6125.png
[20210211015707765.png]: /images/20220829/3cad59090ae04cd3b6177ec7817acdd2.png
[20210211020143946.png]: /images/20220829/4eb6f0dfdad94927a3cddc70d64ddea7.png
[20210211020847745.png]: /images/20220829/970c0dddbf6b4412a8d3ecc8ee3d408d.png
[2021021102181993.png]: /images/20220829/08d1af689b7848b6b28330836421c08b.png
[20210211022059441.png]: /images/20220829/0c16c70d23f34966bd7145eb82a45792.png
[20210211022324362.png]: /images/20220829/c6ff758063cd429ea9286794ee9f15cc.png