Kafka-概述了解、快速入门、部署安装以及基本操作

超、凢脫俗 2022-09-09 00:23 98阅读 0赞

# Kafka 概述 #

Kafka 是一个**分布式**的基于**发布/订阅模式**的**消息队列(Message Queue),主要应用于大数据实时处理**领域，实时数据处理几乎用的都是kafka。

就是消息中间件。

## 消息队列 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]

**使用消息队列的好处**

1)解耦：允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。

2)可恢复性：系统的一部分组件失效时,不会影响到整个系统。消息队列降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。

3)缓冲：有助于控制和优化数据流经过系统的速度,解决生产消息和消费消息的处理速度不一致的情况。主要是处理生产大于消费的情况。

4)灵活性 & 峰值处理能力：在访问量剧增的情况下,应用仍然需要继续发挥作用,但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力,而不会因为突发的超负荷的请求而完全崩溃。消息队列可以动态的增加或减少节点。

5)异步通信：很多时候,用户不想也不需要立即处理消息。消息队列提供了异步处理机制,允许用户把一个消息放入队列,但并不立即处理它。想向队列中放入多少消息就放多少,然后在需要的时候再去处理它们。

### 消息队列的两种模式 ###

(1) **点对点模式**(**一对一**,消费者主动拉取数据,消息收到后消息清除)

消息生产者生产消息发送到 Queue 中,然后消息消费者从 Queue 中取出并且消费消息。消息被消费以后,queue 中不再有存储,所以消息消费者不可能消费到已经被消费的消息。Queue 支持存 在多 个消费 者, 但是对 一个消 息而 言, 只会有 一个消 费者 可以 消费。

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

(2) **发布/订阅模式**(**一对多**,消费者消费数据之后不会清除消息，但是保存消息有时限，到了一定时间之后依旧会删除)

消息生产者(发布)将消息发布到 topic 中,同时有多个消息消费者(订阅)消费该消息。和点对点方式不同,发布到 topic 的消息会被所有订阅者消费。

发布订阅有两种情况，一种是消费者主动拉取（kafka），另一种是队列主动去推数据。前一种会比较浪费资源，因为它需要一直去轮询队列中是否有消息可以消费。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]

## Kafka 基础架构 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 2]

1)**Producer** :消息生产者,就是向 kafka broker 发消息的客户端;

2)**Consumer** :消息消费者,向 kafka broker 取消息的客户端;

3)**Consumer Group (CG)**:消费者组,由多个 consumer 组成。消**费者组内每个消费者负责消费不同分区的数据,一个分区只能由一个组内消费者消费;消费者组之间互不影响**。所有的消费者都属于某个消费者组,**即消费者组是逻辑上的一个订阅者**。用于提高消费能力。当消费者组里消费者的个数与索要消费的topic的分区个数相同时是效率最高的，多了的话多的会浪费，少了的话需要某些消费者消费多个分区。

4)**Broker** :一台 kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker可以容纳多个 topic。

5)**Topic** :可以理解为一个队列,**生产者和消费者面向的都是一个 topic**;订阅了对应的topic才会收到对应的消息

6)**Partition**:为了实现扩展性,一个非常大的 topic 可以分布到多个 broker (即服务器)上,**一个 topic 可以分为多个 partition**,每个 partition 是一个有序的队列;用来负载均衡的，提高对Topic读写的并行度，并且保存消息是通过轮询分区的机制的。

1.  **Replica**:副本,为保证集群中的某个节点发生故障时,**该节点上的 partition 数据不丢失,且 kafka 仍然能够继续工作**, kafka 提供了副本机制,一个 topic 的每个分区都有若干个副本,一个 **leader** 和若干个 **follower**。

8)**leader**:每个分区多个副本的“主”,生产者发送数据的对象,以及消费者消费数据的对象都是 leader。

9)**follower**:每个分区多个副本中的“从”,实时从 leader 中同步数据,保持和 leader 数据的同步。leader 发生故障时,某个 follower 会成为新的 leader 。

1.  **zookeer**:帮助kafka管理整个集群，而kafka集群能够正常的工作也需要依赖zk，zk也会帮助kafka保存一些信息。

# Kafka 快速入门 #

## 安装部署 ##

### 集群规划 ###

每台服务器上均需要安装zk和kafka。kafka是依赖zk的，kafka是scala写的，但是用java操作比较多。

### jar 包下载 ###

http://kafka.apache.org/downloads.html

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 3]

### 集群部署 ###

1)解压安装包
    
    tar -zxvf kafka_2.11-0.11.0.0.tgz -C /opt/module/
    2)修改解压后的文件名称
    mv kafka_2.11-0.11.0.0/ kafka
    
    3)在/opt/module/kafka 目录下创建 data 文件夹
    mkdir data
    
    4)修改配置文件
    cd config/
    vi server.properties
    
    输入以下内容:
    #broker 的全局唯一编号,不能重复
    broker.id=0
    #删除 topic 功能使能，默认false
    delete.topic.enable=true
    #处理网络请求的线程数量
    num.network.threads=3
    #用来处理磁盘 IO 的现成数量
    num.io.threads=8
    #发送套接字的缓冲区大小
    socket.send.buffer.bytes=102400
    #接收套接字的缓冲区大小
    socket.receive.buffer.bytes=102400
    #请求套接字的缓冲区大小
    socket.request.max.bytes=104857600
    #kafka 运行日志存放的路径，也是暂存数据的地方，最好是放在data下，做到日志和数据分离
    log.dirs=/opt/module/kafka/data
    #topic 在当前 broker 上的分区个数
    num.partitions=1
    #用来恢复和清理 data 下数据的线程数量
    num.recovery.threads.per.data.dir=1
    #segment 文件保留的最长时间,超时将被删除
    log.retention.hours=168
    #配置连接 Zookeeper 集群地址
    zookeeper.connect=hadoop113:2181,hadoop114:2181,hadoop115:2181
    
    5)配置环境变量
    sudo vi /etc/profile
    
    #KAFKA_HOME
    export KAFKA_HOME=/opt/module/kafka
    export PATH=$PATH:$KAFKA_HOME/bin
    
    source /etc/profile
    
    6)分发安装包，注意:分发之后记得配置其他机器的环境变量
    xsync kafka/
    
    7)分别在 hadoop114 和 hadoop115 上修改配置件/opt/module/kafka/config/server.properties中的 broker.id=1、broker.id=2，因为broker.id 不得重复
    
    8)启动集群 依次在 hadoop113、hadoop114、hadoop115 节点上启动 kafka
    kafka-server-start.sh -daemon config/server.properties
    
    9)关闭集群
    bin/kafka-server-stop.sh stop

**kafka 群起脚本**

#!/bin/bash
    
    case $1 in
    "start"){
        for i in hadoop113 hadoop114 hadoop115
        do
            echo "========== $i =========="
            ssh $i '/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties'
        done
    };;
    "stop"){
        for i in hadoop113 hadoop114 hadoop115
        do
            echo "========== $i =========="
            ssh $i '/opt/module/kafka/bin/kafka-server-stop.sh /opt/module/kafka/config/server.properties'
        done
    };;
    
    esac

## Kafka 命令行操作 ##

1)查看当前服务器中的所有 topic
    bin/kafka-topics.sh --zookeeper hadoop113:2181 --list
    
    2)创建 topic
    bin/kafka-topics.sh --zookeeper hadoop113:2181 --create --topic first --replication-factor 2 --partitions 2
    Created topic "first".
    三个kafka下有两个的logs目录下有first-0目录，主题名-分区号；logs目录下最重要的日志信息在server.log中
    
    选项说明:
    --topic 定义 topic 名
    --replication-factor 定义副本数，副本数不能大于当前可用的broker数
    --partitions 定义分区数
    
    3)删除 topic
    bin/kafka-topics.sh --zookeeper hadoop113:2181 --delete --topic first
    Topic first is marked for deletion.
    Note: This will have no impact if delete.topic.enable is not set to true.
    需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除
    
    4)查看某个 Topic 的详情
    bin/kafka-topics.sh --zookeeper hadoop113:2181 --describe --topic first
    Topic:first     PartitionCount:2        ReplicationFactor:2     Configs:
            Topic: first    Partition: 0    Leader: 1       Replicas: 1,0   Isr: 1,0
            Topic: first    Partition: 1    Leader: 0       Replicas: 0,1   Isr: 0,1
            
    5)修改分区数
    bin/kafka-topics.sh --zookeeper hadoop113:2181 --alter --topic first --partitions 6

控制台生产者消费者测试：

生产者
    bin/kafka-console-producer.sh --broker-list hadoop113:9092 --topic first
    
    消费者方法一
    bin/kafka-console-consumer.sh --zookeeper hadoop113:2181 --topic first
    
    消费者方法二
    bin/kafka-console-consumer.sh --bootstrap-server hadoop113:9092 --topic first
    
    消费者方法三
    --from-beginning:会把主题中以往所有的数据都读取出来
    bin/kafka-console-consumer.sh --bootstrap-server hadoop113:9092 --from-beginning --topic first
    
    消费者消费完之后会将消费的偏移量根据分区保存在消费这个本机的logs目录下（以轮询的方式保存），默认是50个分区一个副本：
    hadoop113：
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-0
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-10
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-12
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-14
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-16
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-18
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-2
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-20
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-22
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-24
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-26
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-28
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-30
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-32
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-34
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-36
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-38
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-4
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-40
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-42
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-44
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-46
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-48
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-6
    drwxrwxr-x 2 bd bd    141 8月  26 16:34 __consumer_offsets-8

数据日志分离，先停止zk集群以及kk集群，然后删除kk的logs和/opt/module/zookeeper-3.5.7/zkData/version-2（这一步相当于初始化zk，因为version-2是保存zk数据的地方）

将各个kk配置文件 server.properties中的log.dirs=/opt/module/kafka/logs改为log.dirs=/opt/module/kafka/data

然后重启zk和kk即做到了kk的数据和日志分离，数据在data下，日志在logs下。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/525ce6ab7b8d467ca12c76553d58ba63.png
[61ab486cd8cb47aea386b608e610489e.png]: /images/20220829/89d2b751e30d4bf39761d6c07685999d.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/c8cec15df0db412593d680f58958cbde.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/64f98e1d23c44cf7a7ba25a1db0725ca.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/8e15e01b2f414203ae090da325754485.png