简单的kafkaredis学习之kafka

简单的kafka&redis学习整理之kafka

1. kafka

1.1 什么是消息队列

在学习Kafka之前我们先来看一下什么是消息队列，消息队列(Message Queue)：可以简称为MQ

例如：Java中的Queue队列，也可以认为是一个消息队列

消息队列：顾名思义，消息+队列，其实就是保存消息的队列，属于消息传输过程中的容器。消息队列主要提供生产、消费接口供外部调用，做数据的存储和读取

1.2 消息队列分类

消息队列大致可以分为两种：点对点(P2P)、发布订阅(Pub/Sub)

• 共同点：
  针对数据的处理流程是一样的
  消息生产者生产消息发送到queue中，然后消息消费者从queue中读取并且消费消息。
• 不同点：
  点对点(p2p)模型包含：消息队列(Queue)、发送者(Sender)、接收者(Receiver)
  一个生产者生产的消息只有一个消费者(Consumer)（消息一旦被消费，就不在消息队列中）消费。例如QQ中的私聊，我发给你的消息只有你能看到，别人是看不到的

发布订阅(Pub/Sub)模型包含：消息队列(Queue)、主题（Topic）、发布者（Publisher）、订阅者（Subscriber）
每个消息可以有多个消费者，彼此互不影响。比如我发布一个微博：关注我的人都能够看到，或者QQ中的群聊，我在群里面发一条消息，群里面所有人都能看到

这就是这两种消息队列的区别，我们接下来要学习的Kafka这个消息队列是属于发布订阅模型的

1.3 什么是Kafka

Kafka 是一个 高吞吐量 的、 持久性 的、 分布式 发布订阅消息系统

• 高吞吐量：可以满足每秒百万级别消息的生产和消费。
  为什么这么快？难道Kafka的数据是放在内存里面的吗？不是的，Kafka的数据还是放在磁盘里面的，主要是Kafka利用了磁盘顺序读写速度超过内存随机读写速度这个特性。所以说它的吞吐量才这么高
• 持久性：有一套完善的消息存储机制，确保数据高效安全的持久化。
• 分布式：它是基于分布式的扩展、和容错机制；Kafka的数据都会复制到几台服务器上。当某一台机器故障失效时，生产者和消费者切换使用其它的机器。

Kafka的数据时存储是磁盘中的，为什么可以满足每秒百万级别消息的生产和消费？
这是一个面试题，其实就是我们刚才针对高吞吐量的解释：kafka利用了磁盘顺序读写速度超过内存随机读写速度这个特性

Kafka主要应用在实时计算领域，可以和Flume、Spark、Flink等框架结合在一块使用

例如：我们使用Flume采集网站产生的日志数据，将数据写入到Kafka中，然后通过Spark或者Flink从Kafka中消费数据进行计算，这其实是一个典型的实时计算案例的架构

1.4 Kafka组件介绍

接下来我们来分析一下Kafka中的组件，加深对kafka的理解，看这个图

• 先看中间的Kafka Cluster
  这个Kafka集群内有两个节点，这些节点在这里我们称之为Broker
  Broker：消息的代理，Kafka集群中的一个节点称为一个broker

• 在Kafka中有Topic的概念
  Topic：称为主题，Kafka处理的消息的不同分类(是一个逻辑概念)。
  如果把Kafka认为是一个数据库的话，那么Kafka中的Topic就可以认为是一张表
  不同的topic中存储不同业务类型的数据，方便使用

• 在Topic内部有partition的概念
  Partition：是Topic物理上的分组，一个Topic会被分为1个或者多个partition(分区)，分区个数是在创建topic的时候指定。每个topic都是有分区的，至少1个。
  注意：这里面针对partition其实还有副本的概念，主要是为了提供数据的容错性，我们可以在创建Topic的时候指定partition的副本因子是几个。
  在这里面副本因子其实就是2了，其中一个是Leader，另一个是真正的副本
  Leader中的这个partition负责接收用户的读写请求，副本partition负责从Leader里面的partiton中同步数据，这样的话，如果后期leader对应的节点宕机了，副本可以切换为leader顶上来。

• 在partition内部还有一个message的概念
  Message：我们称之为消息，代表的就是一条数据，它是通信的基本单位，每个消息都属于一个partition。

在这里总结一下：
Broker>Topic>Partition>Message，接下来还有两个组件，看图中的最左边和最右边

• Producer：消息和数据的生产者，向Kafka的topic生产数据。
• Consumer：消息和数据的消费者，从kafka的topic中消费数据。
  这里的消费者可以有多个，每个消费者可以消费到相同的数据

最后还有一个Zookeeper服务，Kafka的运行是需要依赖于Zookeeper的，Zookeeper负责协调Kafka集群的正常运行。

1.5 kafka安装使用

前面我们对Kafka有了一个基本的认识，下面我们就想使用一下Kafka，在使用之前，需要先把Kafka安装部署起来。Kafka是支持单机和集群模式的，建议大家在学习阶段使用单机模式即可，单机和集群在操作上没有任何区别

注意：由于Kafka需要依赖于Zookeeper，所以在这我们需要先把Zookeeper安装部署起来

1.5.1 zookeeper安装部署

针对Zookeeper前期不需要掌握太多，只需要掌握Zookeeper的安装部署以及它的基本操作即可。

Zookeeper也支持单机和集群安装，建议大家在学习阶段使用单机即可，单机和集群在操作上没有任何区别。在这里我们会针对单机和集群这两种方式都演示一下

• zookeeper单机安装

  1：zookeeper需要依赖于jdk，只要保证jdk已经正常安装即可。具体安装步骤如下，下载zookeeper的安装包进入Zookeeper的官网

  

  ​

  最终下载链接如下：https://mirrors.bfsu.edu.cn/apache/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gz

  2：把安装包上传到bigdata01机器的/data/soft目录下

  [root@bigdata01 soft]# ll
  -rw-r--r--.  1 root root    9394700 Jun  2 21:33 apache-zookeeper-3.5.8-bin.tar.gz
  

  解压安装包

  [root@bigdata01 soft]# tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
  

  3：修改配置文件
  首先将zoo_sample.cfg重命名为zoo.cfg
  然后修改zoo.cfg中的dataDir参数的值，dataDir指向的目录存储的是zookeeper的核心数据，所以这个目录不能使用tmp目录

  [root@bigdata01 soft]# cd apache-zookeeper-3.5.8-bin/conf
  [root@bigdata01 conf]# mv zoo_sample.cfg  zoo.cfg
  [root@bigdata01 conf]# vi zoo.cfg 
  dataDir=/data/soft/apache-zookeeper-3.5.8-bin/data
  

  4：启动zookeeper服务

  [root@bigdata01 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh start
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Starting zookeeper ... STARTED
  

  5：验证
  如果能看到QuorumPeerMain进程就说明zookeeper启动成功

  [root@bigdata01 apache-zookeeper-3.5.8-bin]# jps
  1701 QuorumPeerMain
  1733 Jps
  

  注意：如果执行jps命令发现没有QuorumPeerMain进程，则需要到logs目录下去查看zookeeper-*.out这个日志文件

  也可以通过zkServer.sh 脚本查看当前机器中zookeeper服务的状态

  注意：使用zkServer.sh默认会连接本机2181端口的zookeeper服务，默认情况下zookeeper会监听2181端口，这个需要注意一下，因为后面我们在使用zookeeper的时候需要知道它监听的端口是哪个。

  最下面显示的Mode信息，表示当前是一个单机独立集群

  [root@bigdata01 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh status
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Client port found: 2181. Client address: localhost.
  Mode: standalone
  

  如果没有启动成功的话则会提示连不上服务not running

  6：操作zookeeper

  首先使用zookeeper的客户端工具连接到zookeeper里面，使用bin目录下面的zKCli.sh脚本，默认会连接本机的zookeeper服务

  [root@bigdata01 apache-zookeeper-3.5.8-bin]# bin/zkCli.sh
  Connecting to localhost:2181
  .....
  WATCHER::
  
  WatchedEvent state:SyncConnected type:None path:null
  [zk: localhost:2181(CONNECTED) 0] 
  

  这样就进入zookeeper的命令行了。在这里面可以操作Zookeeper中的目录结构，zookeeper中的目录结构和Linux文件系统的目录结构类似，zookeeper里面的每一个目录我们称之为节点(ZNode)。

  正常情况下我们可以把ZNode认为和文件系统中的目录类似，但是有一点需要注意：ZNode节点本身是可以存储数据的。

  zookeeper中提供了一些命令可以对它进行一些操作，在命令行下随便输入一个字符，按回车就会提示出zookeeper支持的所有命令

  [zk: localhost:2181(CONNECTED) 8] aa
  ZooKeeper -server host:port cmd args
          addauth scheme auth
          close 
          config [-c] [-w] [-s]
          connect host:port
          create [-s] [-e] [-c] [-t ttl] path [data] [acl]
          delete [-v version] path
          deleteall path
          delquota [-n|-b] path
          get [-s] [-w] path
          getAcl [-s] path
          history 
          listquota path
          ls [-s] [-w] [-R] path
          ls2 path [watch]
          printwatches on|off
          quit 
          reconfig [-s] [-v version] [[-file path] | [-members serverID=host:port1:port2;port3[,...]*]] | [-add serverId=host:port1:port2;port3[,...]]* [-remove serverId[,...]*]
          redo cmdno
          removewatches path [-c|-d|-a] [-l]
          rmr path
          set [-s] [-v version] path data
          setAcl [-s] [-v version] [-R] path acl
          setquota -n|-b val path
          stat [-w] path
          sync path
  Command not found: Command not found aa
  

  下面我们来具体看一些比较常用的功能：

  • 查看根节点下面有什么内容

    这里显示根节点下面有一个zookeeper节点

    [zk: localhost:2181(CONNECTED) 0] ls /
    [zookeeper]
    

  • 创建节点

    在根节点下面创建一个test节点，在test节点上存储数据hello

    [zk: localhost:2181(CONNECTED) 9] create /test hello
    Created /test
    

  • 查看节点中的信息

    查看/test节点中的内容

    [zk: localhost:2181(CONNECTED) 10] get /test
    hello
    

  • 删除节点

    这个删除命令可以递归删除，这里面还有一个delete命令，也可以删除节点，但是只能删除空节点，如果节点下面还有子节点，想一次性全部删除建议使用deleteall

    [zk: localhost:2181(CONNECTED) 6] deleteall /test
    

  直接按ctrl+c就可以退出这个操作界面，想优雅一些的话可以输入quit退出

  7：停止zookeeper服务

  [root@bigdata01 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh stop
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Stopping zookeeper ... STOPPED
  

• zookeeper集群安装

  1：集群节点规划，使用三个节点搭建一个zookeeper集群

  bigdata01
  bigdata02
  bigdata03
  

  2：首先在bigdata01节点上配置zookeeper
  解压

  [root@bigdata01 soft]# tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
  

  修改配置，将zoo_sample.cfg重命名为zoo.cfg，然后修改zoo.cfg中的dataDir参数的值，dataDir指向的目录存储的是zookeeper的核心数据，所以这个目录不能使用tmp目录，然后增加server.0、server.1、server.2这三行内容

  [root@bigdata01 soft]# cd apache-zookeeper-3.5.8-bin/conf/
  [root@bigdata01 conf]# mv zoo_sample.cfg  zoo.cfg
  [root@bigdata01 conf]# vi zoo.cfg
  dataDir=/data/soft/apache-zookeeper-3.5.8-bin/data
  server.0=bigdata01:2888:3888
  server.1=bigdata02:2888:3888
  server.2=bigdata03:2888:3888
  

  创建目录保存myid文件，并且向myid文件中写入内容，myid中的值其实是和zoo.cfg中server后面指定的编号是一一对应的，编号0对应的是bigdata01这台机器，所以在这里指定0，在这里使用echo 和 重定向 实现数据写入

  [root@bigdata01 conf]#cd /data/soft/apache-zookeeper-3.5.8-bin
  [root@bigdata01 apache-zookeeper-3.5.8-bin]# mkdir data
  [root@bigdata01 apache-zookeeper-3.5.8-bin]# cd data
  [root@bigdata01 data]# echo 0 > myid 
  

  3：把修改好配置的zookeeper拷贝到其它两个节点

  [root@bigdata01 soft]# scp -rq apache-zookeeper-3.5.8-bin bigdata02:/data/soft/
  [root@bigdata01 soft]# scp -rq apache-zookeeper-3.5.8-bin bigdata03:/data/soft/
  

  4：修改bigdata02和bigdata03上zookeeper中myid文件的内容
  首先修改bigdata02节点上的myid文件

  [root@bigdata02 ~]# cd /data/soft/apache-zookeeper-3.5.8-bin/data/
  [root@bigdata02 data]# echo 1 > myid
  

  然后修改bigdata03节点上的myid文件

  [root@bigdata03 ~]# cd /data/soft/apache-zookeeper-3.5.8-bin/data/
  [root@bigdata03 data]# echo 2 > myid
  

  5：启动zookeeper集群
  分别在bigdata01、bigdata02、bigdata03上启动zookeeper进程
  在bigdata01上启动

  [root@bigdata01 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh start
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Starting zookeeper ... STARTED
  

  在bigdata02上启动

  [root@bigdata02 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh start
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Starting zookeeper ... STARTED
  

  在bigdata03上启动

  [root@bigdata03 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh start
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Starting zookeeper ... STARTED
  

  6：验证
  分别在bigdata01、bigdata02、bigdata03上执行jps命令验证是否有QuorumPeerMain进程
  如果都有就说明zookeeper集群启动正常了
  如果没有就到对应的节点的logs目录下查看zookeeper*-*.out日志文件

  执行bin/zkServer.sh status 命令会发现有一个节点显示为leader，其他两个节点为follower

  [root@bigdata01 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh status
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Client port found: 2181. Client address: localhost.
  Mode: follower
  
  [root@bigdata02 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh status
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Client port found: 2181. Client address: localhost.
  Mode: leader
  
  [root@bigdata03 apache-zookeeper-3.5.8-bin]# bin/zkServer.sh status
  ZooKeeper JMX enabled by default
  Using config: /data/soft/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg
  Client port found: 2181. Client address: localhost.
  Mode: follower
  

  7：操作zookeeper,和上面单机的操作方式一样

  8：停止zookeeper集群,在bigdata01、bigdata02、bigdata03三台机器上分别执行bin/zkServer.sh stop命令

1.5.2 kafka安装部署

zookeeper集群安装好了以后就可以开始安装kafka了。在安装kafka之前需要先确保zookeeper集群是启动状态。kafka还需要依赖于基础环境jdk，需要确保jdk已经安装到位。

• kafka单机安装

  1：下载kafka安装包

  

  最终下载链接如下：https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.1/kafka_2.12-2.4.1.tgz

  注意：kafka在启动的时候不需要安装scala环境，只有在编译源码的时候才需要，因为运行的时候是在jvm虚拟机上运行的，只需要有jdk环境就可以了

  2：把kafka安装包上传到bigdata01的/data/soft目录下

  [root@bigdata01 soft]# ll
  -rw-r--r--.  1 root root   62358954 Jun  2 21:22 kafka_2.12-2.4.1.tgz
  

  3：解压

  [root@bigdata01 kafka_2.12-2.4.1]# tar -zxvf kafka_2.12-2.4.1.tgz
  

  4：修改配置文件
  主要参数：

  broker.id：集群节点id编号，单机模式不用修改
  listeners：默认监听9092端口
  log.dirs：注意：这个目录不是存储日志的，是存储Kafka中核心数据的目录，这个目录默认是指向的tmp目录，所以建议修改一下
  zookeeper.connect：kafka依赖的zookeeper
  

  单机模式，如果kafka和zookeeper在同一台机器上，zookeeper监听的端口就是那个默认的2181端口，则zookeeper.connect这个参数就不需要修改了。
  只需要修改一下log.dirs即可

  [root@bigdata01 kafka_2.12-2.4.1]# cd kafka_2.12-2.4.1/config/
  [root@bigdata01 config]# vi server.properties
  log.dirs=/data/soft/kafka_2.12-2.4.1/kafka-logs
  

  5：启动kafka

  [root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-server-start.sh -daemon config/server.properties 
  

  6：验证
  启动成功之后会产生一个kafka进程

  [root@bigdata01 kafka_2.12-2.4.1]# jps
  2230 QuorumPeerMain
  3117 Kafka
  3182 Jps
  

  7：停止kafka

  [root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-server-stop.sh 
  

• kafka集群安装

  1：集群节点规划，使用三个节点搭建一个kafka集群

  bigdata01
  bigdata02
  bigdata03
  

  注意：针对Kafka集群而言，没有主从之分，所有节点都是一样的。
  2：首先在bigdata01节点上配置kafka

  解压：

  [root@bigdata01 soft]# tar -zxvf kafka_2.12-2.4.1.tgz
  

  修改配置文件

  注意：此时针对集群模式需要修改 broker.id、log.dirs、以及zookeeper.connect

  broker.id的值默认是从0开始的，集群中所有节点的broker.id从0开始递增即可，所以bigdata01节点的broker.id值为0
  log.dirs的值建议指定到一块存储空间比较大的磁盘上面，因为在实际工作中kafka中会存储很多数据，我这个虚拟机里面就一块磁盘，所以就指定到/data目录下面了
  zookeeper.connect的值是zookeeper集群的地址，可以指定集群中的一个节点或者多个节点地址，多个节点地址之间使用逗号隔开即可

  [root@bigdata01 soft]# cd kafka_2.12-2.4.1/config/
  [root@bigdata01 config]# vi server.properties
  broker.id=0
  log.dirs=/data/kafka-logs
  zookeeper.connect=bigdata01:2181,bigdata02:2181,bigdata03:2181
  

  3：将修改好配置的kafka安装包拷贝到其它两个节点

  [root@bigdata01 soft]# scp -rq kafka_2.12-2.4.1 bigdata02:/data/soft/
  [root@bigdata01 soft]# scp -rq kafka_2.12-2.4.1 bigdata03:/data/soft/
  

  4：修改bigdata02和bigdata03上kafka中broker.id的值
  首先修改bigdata02节点上的broker.id的值为1

  [root@bigdata02 ~]# cd /data/soft/kafka_2.12-2.4.1/config/
  [root@bigdata02 config]# vi server.properties 
  broker.id=1
  

  然后修改bigdata03节点上的broker.id的值为2

  [root@bigdata03 ~]# cd /data/soft/kafka_2.12-2.4.1/config/
  [root@bigdata03 config]# vi server.properties 
  broker.id=2
  

  5：启动集群
  分别在bigdata01、bigdata02、bigdata03上启动kafka进程在bigdata01上启动

  [root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-server-start.sh -daemon config/server.properties 
  

  在bigdata02上启动

  [root@bigdata02 kafka_2.12-2.4.1]# bin/kafka-server-start.sh -daemon config/server.properties 
  

  在bigdata03上启动

  [root@bigdata03 kafka_2.12-2.4.1]# bin/kafka-server-start.sh -daemon config/server.properties 
  

  6：验证
  分别在bigdata01、bigdata02、bigdata03上执行jps命令验证是否有kafka进程，如果都有就说明kafka集群启动正常了

1.5.3 Kafka中Topic的操作

kafka集群安装好了以后我们就想向kafka中添加一些数据，想要添加数据首先需要创建topic，那接下来看一下针对topic的一些操作

新增Topic：指定2个分区，2个副本，注意：副本数不能大于集群中Broker的数量

因为每个partition的副本必须保存在不同的broker，否则没有意义，如果partition的副本都保存在同一个broker，那么这个broker挂了，则partition数据依然会丢失，在这里我使用的是3个节点的kafka集群，所以副本数我就暂时设置为2，最大可以设置为3。

如果你们用的是单机kafka的话，这里的副本数就只能设置为1了，这个需要注意一下

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --create --zookeeper localhost:2181  --partitions 1 --replication-factor 1 --topic hello
Created topic hello.

查询Topic：查询kafka中所有的topic列表

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --list --zookeeper localhost:2181
hello

查看指定topic的详细信息

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic hello
Topic: hello    PartitionCount: 2       ReplicationFactor: 2    Configs: 
        Topic: hello    Partition: 0    Leader: 2       Replicas: 2,0      Isr: 2,0
        Topic: hello    Partition: 1    Leader: 0       Replicas: 0,1      Isr: 0,1

第一个行显示指定topic所有partitions的一个总结

• PartitionCount：表示这个Topic一共有多少个partition
• ReplicationFactor：表示这个topic中partition的副本因子是几
• Config：这个表示创建Topic时动态指定的配置信息，在这我们没有额外指定配置信息

下面每一行给出的是一个partition的信息，如果只有一个partition，则只显示一行。

• Topic：显示当前的topic名称
• Partition：显示当前topic的partition编号
• Leader：Leader partition所在的节点编号，这个编号其实就是broker.id的值，

来看这个图：

这个图里面的hello这个topic有两个partition，其中partition1的leader所在的节点是broker1，partition2的leader所在的节点是broker2

Replicas：当前partition所有副本所在的节点编号【包含Leader所在的节点】，如果设置多个副本的话，这里会显示多个，不管该节点是否是Leader以及是否存活。
Isr：当前partition处于同步状态的所有节点，这里显示的所有节点都是存活状态的，并且跟Leader同步的(包含Leader所在的节点)

所以说Replicas和Isr的区别就是
如果某个partition的副本所在的节点宕机了，在Replicas中还是会显示那个节点，但是在Isr中就不会显示了，Isr中显示的都是处于正常状态的节点。

• 修改Topic：修改Topic的partition数量，只能增加

  为什么partition只能增加？因为数据是存储在partition中的，如果可以减少partition的话，那么partition中的数据就丢了

  [root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --alter --zookeeper localhost:2181 --partitions 5 --topic hello
  WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
  Adding partitions succeeded!
  

  修改之后再来查看一下topic的详细信息

  [root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic hello
  Topic: hello    PartitionCount: 5       ReplicationFactor: 2    Configs: 
          Topic: hello    Partition: 0    Leader: 2       Replicas: 2,0       Isr: 2,0
          Topic: hello    Partition: 1    Leader: 0       Replicas: 0,1       Isr: 0,1
          Topic: hello    Partition: 2    Leader: 1       Replicas: 1,2       Isr: 1,2
          Topic: hello    Partition: 3    Leader: 2       Replicas: 2,1       Isr: 2,1
          Topic: hello    Partition: 4    Leader: 0       Replicas: 0,2       Isr: 0,2
  

• 删除Topic：删除Kafka中的指定Topic

  [root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic hello
  Topic hello is marked for deletion.
  Note: This will have no impact if delete.topic.enable is not set to true.
  

  删除操作是不可逆的，删除Topic会删除它里面的所有数据

  注意：Kafka从1.0.0开始默认开启了删除操作，之前的版本只会把Topic标记为删除状态，需要设置delete.topic.enable为true才可以真正删除

  如果不想开启删除功能，可以设置delete.topic.enable为false，这样删除topic的时候只会把它标记为删除状态，此时这个topic依然可以正常使用。
  delete.topic.enable可以配置在server.properties文件中

  

1.5.4 Kafka中的生产者和消费者

前面我们学习了Kafka中的topic的创建方式，下面我们可以向topic中生产数据以及消费数据了
生产数据需要用到生产者
消费数据需要用到消费者

kafka默认提供了基于控制台的生产者和消费者，方便测试使用

• 生产者：bin/kafka-console-producer.sh
• 消费者：bin/kafka-console-consumer.sh

先来看一下如何向里面生产数据，直接使用kafka提供的基于控制台的生产者
先创建一个topic【5个分区，2个副本】：

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --create --zookeeper localhost:2181  --partitions 5 --replication-factor 2 --topic hello 
Created topic hello.

向这个topic中生产数据
broker-list：kafka的服务地址[多个用逗号隔开]、
topic：topic名称

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic hello
>hehe

下面来创建一个消费者消费topic中的数据，bootstrap-server：kafka的服务地址，topic:具体的topic

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello

发现消费不到刚才生产的数据，为什么呢？
因为kafka的消费者默认是消费最新生产的数据，如果想消费之前生产的数据需要添加一个参数–from-beginning，表示从头消费的意思

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello --from-beginning
hehe

1.5.5 案例：QQ群聊天

通过kafka可以模拟QQ群聊天的功能，我们来看一下
首先在kafka中创建一个新的topic，可以认为是我们在QQ里面创建了一个群，群号是88888888

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --create --zookeeper localhost:2181  --partitions 5 --replication-factor 2 --topic 88888888
Created topic 88888888.

然后我把你们都拉到这个群里面，这样我在群里面发消息你们就都能收到了
在bigdata02和bigdata03上开启消费者，可以认为是把这两个人拉到群里面了

[root@bigdata02 kafka_2.12-2.4.1]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic 88888888
[root@bigdata03 kafka_2.12-2.4.1]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic 88888888

然后我在bigdata01上开启生产者发消息，这样bigdata02和bigdata03都是可以收到的。
这样就可以认为在群里的人都能收到我发的消息，类似于发广播。
这个其实主要利用了kafka中的多消费者的特性，每个消费者都可以消费到相同的数据

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic 88888888
>hello everyone

1.6 kafka核心扩展

1.6.1 Broker扩展

Broker的参数可以配置在server.properties这个配置文件中，Broker中支持的完整参数在官方文档中有体现

具体链接为：http://kafka.apache.org/24/documentation.html#brokerconfigs

针对Broker的参数，我们主要分析两块
1：Log Flush Policy：设置数据flush到磁盘的时机
为了减少磁盘写入的次数,broker会将消息暂时缓存起来,当消息的个数达到一定阀值或者过了一定的时间间隔后,再flush到磁盘,这样可以减少磁盘IO调用的次数。
这块主要通过两个参数控制

• log.flush.interval.messages 一个分区的消息数阀值，达到该阈值则将该分区的数据flush到磁盘，注意这里是针对分区，因为topic是一个逻辑概念，分区是真实存在的，每个分区会在磁盘上产生一个目录

  [root@bigdata01 kafka-logs]# ll
  total 20
  drwxr-xr-x. 2 root root 141 Jun  8 17:12 88888888-0
  drwxr-xr-x. 2 root root 141 Jun  8 17:12 88888888-3
  -rw-r--r--. 1 root root   4 Jun  8 15:23 cleaner-offset-checkpoint
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-0
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-12
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-15
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-18
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-21
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-24
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-27
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-3
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-30
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-33
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-36
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-39
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-42
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-45
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-48
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-6
  drwxr-xr-x. 2 root root 141 Jun  8 17:08 __consumer_offsets-9
  drwxr-xr-x. 2 root root 141 Jun  8 17:04 hello-1
  drwxr-xr-x. 2 root root 141 Jun  8 17:04 hello-4
  

  这个参数的默认值为9223372036854775807，long的最大值
  默认值太大了，所以建议修改，可以使用server.properties中针对这个参数指定的值10000，需要去掉注释之后这个参数才生效。

• log.flush.interval.ms 间隔指定时间
  默认间隔指定的时间将内存中缓存的数据flush到磁盘中，由文档可知，这个参数的默认值为null，此时会使用log.flush.scheduler.interval.ms参数的值，log.flush.scheduler.interval.ms参数的值默认是9223372036854775807，long的最大值

  所以这个值也建议修改，可以使用server.properties中针对这个参数指定的值1000，单位是毫秒，表示每1秒写一次磁盘，这个参数也需要去掉注释之后才生效

2：Log Retention Policy：设置数据保存周期，默认7天
kafka中的数据默认会保存7天，如果kafka每天接收的数据量过大，这样是很占磁盘空间的，建议修改数据保存周期，我们之前在实际工作中是将数据保存周期改为了1天。

数据保存周期主要通过这几个参数控制

• log.retention.hours，这个参数默认值为168，单位是小时，就是7天，可以在这调整数据保存的时间，超过这个时间数据会被自动删除
• log.retention.bytes，这个参数表示当分区的文件达到一定大小的时候会删除它，如果设置了按照指定周期删除数据文件，这个参数不设置也可以，这个参数默认是没有开启的
• log.retention.check.interval.ms，这个参数表示检测的间隔时间，单位是毫秒，默认值是300000，就是5分钟，表示每5分钟检测一次文件看是否满足删除的时机

1.6.2 Producer扩展

Producer默认是随机将数据发送到topic的不同分区中，也可以根据用户设置的算法来根据消息的key来计算输入到哪个partition里面
此时需要通过partitioner来控制，这个知道就行了，因为在实际工作中一般在向kafka中生产数据的都是不带key的，只有数据内容，所以一般都是使用随机的方式发送数据

在这里有一个需要注意的内容就是，针对producer的数据通讯方式：同步发送和异步发送

• 同步是指：生产者发出数据后，等接收方发回响应以后再发送下个数据的通讯方式。
• 异步是指：生产者发出数据后，不等接收方发回响应，接着发送下个数据的通讯方式。

具体的数据通讯策略是由acks参数控制的
acks默认为1，表示需要Leader节点回复收到消息，这样生产者才会发送下一条数据
acks：all，表示需要所有Leader+副本节点回复收到消息（acks=-1），这样生产者才会发送下一条数据
acks：0，表示不需要任何节点回复，生产者会继续发送下一条数据

再来看一下这个图：

我们在向hello这个topic生产数据的时候，可以在生产者中设置acks参数，
acks设置为1，表示我们在向hello这个topic的partition1这个分区写数据的时候，只需要让leader所在的broker1这个节点回复确认收到的消息就可以了，这样生产者就可以发送下一条数据了

如果acks设置为all，则需要partition1的这两个副本所在的节点(包含Leader)都回复收到消息，生产者才会发送下一条数据

如果acks设置为0，表示生产者不会等待任何partition所在节点的回复，它只管发送数据，不管你有没有收到，所以这种情况丢失数据的概率比较高。

针对这块在面试的时候会有一个面试题：Kafka如何保证数据不丢？

其实就是通过acks机制保证的，如果设置acks为all，则可以保证数据不丢，因为此时把数据发送给kafka之后，会等待对应partition所在的所有leader和副本节点都确认收到消息之后才会认为数据发送成功了，所以在这种策略下，只要把数据发送给kafka之后就不会丢了。
如果acks设置为1，则当我们把数据发送给partition之后，partition的leader节点也确认收到了，但是leader回复完确认消息之后，leader对应的节点就宕机了，副本partition还没来得及将数据同步过去，所以会存在丢失的可能性。
不过如果宕机的是副本partition所在的节点，则数据是不会丢的

如果acks设置为0的话就表示是顺其自然了，只管发送，不管kafka有没有收到，这种情况表示对数据丢不丢都无所谓了。

1.6.3 Consumer扩展

在消费者中还有一个消费者组的概念，每个consumer属于一个消费者组，通过group.id指定消费者组，那组内消费和组间消费有什么区别吗？

• 组内：消费者组内的所有消费者消费同一份数据；

  注意：在同一个消费者组中，一个partition同时只能有一个消费者消费数据，如果消费者的个数小于分区的个数，一个消费者会消费多个分区的数据。
  如果消费者的个数大于分区的个数，则多余的消费者不消费数据

  所以，对于一个topic,同一个消费者组中推荐不能有多于分区个数的消费者,否则将意味着某些消费者将无法获得消息。

• 组间：多个消费者组消费相同的数据，互不影响。

  来看下面这个图，加深一下理解

  

Kafka集群有两个节点，Broker1和Broker2，集群内有一个topic，这个topic有4个分区，P0,P1,P2,P3。下面有两个消费者组Consumer Group A和Consumer Group B。其中Consumer Group A中有两个消费者 C1和C2，由于这个topic有4个分区，所以，C1负责消费两个分区的数据，C2负责消费两个分区的数据，这个属于组内消费

Consumer Group B有5个消费者，C3~C7，其中C3,C4,C5,C6分别消费一个分区的数据，而C7就是多余出来的了，因为现在这个消费者组内的消费者的数量比对应的topic的分区数量还多，但是一个分区同时只能被一个消费者消费，所以就会有一个消费者处于空闲状态。
这个也属于组内消费

Consumer Group A和Consumer Group B这两个消费者组属于组间消费，互不影响。

1.6.4 Topic、Partition扩展

每个partition在存储层面是append log文件。新消息都会被直接追加到log文件的尾部，每条消息在log文件中的位置称为offset(偏移量)。越多partitions可以容纳更多的consumer,有效提升并发消费的能力。

具体什么时候增加topic的数量？什么时候增加partition的数量呢？

业务类型增加需要增加topic、数据量大需要增加partition

1.6.5 Message扩展

每条Message包含了以下三个属性：

1. offset 对应类型：long 表示此消息在一个partition中的起始的位置。可以认为offset是partition中Message的id，自增的
2. MessageSize 对应类型：int32 此消息的字节大小。
3. data 是message的具体内容。

看这个图，加深对Topic、Partition、Message的理解

1.6.6 存储策略

在kafka中每个topic包含1到多个partition，每个partition存储一部分Message。每条Message包含三个属性，其中有一个是offset。

问题来了：offset相当于partition中这个message的唯一id，那么如何通过id高效的找到message？

两大法宝：分段+索引

kafak中数据的存储方式是这样的：

1. 每个partition由多个segment【片段】组成，每个segment中存储多条消息，
2. 每个partition在内存中对应一个index，记录每个segment中的第一条消息偏移量。

Kafka中数据的存储流程是这样的，生产者生产的消息会被发送到topic的多个partition上，topic收到消息后往对应partition的最后一个segment上添加该消息，segment达到一定的大小后会创建新的segment。

来看这个图，可以认为是针对topic中某个partition的描述

图中左侧就是索引，右边是segment文件，左边的索引里面会存储每一个segment文件中第一条消息的偏移量，由于消息的偏移量都是递增的，这样后期查找起来就方便了，先到索引中判断数据在哪个segment文件中，然后就可以直接定位到具体的segment文件了，这样再找具体的那一条数据就很快了，因为都是有序的。

1.6.7 容错机制

当Kafka集群中的一个Broker节点宕机，会出现什么现象？下面来演示一下，使用kill -9 杀掉bigdata01中的broker进程测试

[root@bigdata01 kafka_2.12-2.4.1]# jps
7522 Jps
2054 Kafka
1679 QuorumPeerMain
[root@bigdata01 kafka_2.12-2.4.1]# kill 2054

我们可以先通过zookeeper来查看一下，因为当kafka集群中的broker节点启动之后，会自动向zookeeper中进行注册，保存当前节点信息

[root@bigdata01 apache-zookeeper-3.5.8-bin]# bin/zkCli.sh 
Connecting to localhost:2181
.....
[zk: localhost:2181(CONNECTED) 1] ls /brokers
[ids, seqid, topics]
[zk: localhost:2181(CONNECTED) 2] ls /brokers/ids
[1, 2]

此时发现zookeeper的/brokers/ids下面只有2个节点信息，可以通过get命令查看节点信息，这里面会显示对应的主机名和端口号

[zk: localhost:2181(CONNECTED) 4] get /brokers/ids/1
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://bigdata02:9092"],"jmx_port":-1,"host":"bigdata02","timestamp":"1591668875280","port":9092,"version":4}

然后再使用describe查询topic的详细信息，会发现此时的分区的leader全部变成了目前存活的另外两个节点，此时可以发现Isr中的内容和Replicas中的不一样了，因为Isr中显示的是目前正常运行的节点。

所以当Kafka集群中的一个Broker节点宕机之后，对整个集群而言没有什么特别的大影响，此时集群会给partition重新选出来一些新的Leader节点

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic hello                       
Topic: hello    PartitionCount: 5       ReplicationFactor: 2    Configs: 
        Topic: hello    Partition: 0    Leader: 2       Replicas: 2,0       Isr: 2
        Topic: hello    Partition: 1    Leader: 1       Replicas: 0,1       Isr: 1
        Topic: hello    Partition: 2    Leader: 1       Replicas: 1,2       Isr: 1,2
        Topic: hello    Partition: 3    Leader: 2       Replicas: 2,1       Isr: 2,1
        Topic: hello    Partition: 4    Leader: 2       Replicas: 0,2       Isr: 2

当Kafka集群中新增一个Broker节点，会出现什么现象？新加入一个broker节点，zookeeper会自动识别并在适当的机会选择此节点提供服务，再次启动bigdata01节点中的broker进程测试

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-server-start.sh -daemon config/server.properties 

此时到zookeeper中查看一下

[root@bigdata01 apache-zookeeper-3.5.8-bin]# bin/zkCli.sh 
Connecting to localhost:2181
.....
[zk: localhost:2181(CONNECTED) 1] ls /brokers
[ids, seqid, topics]
[zk: localhost:2181(CONNECTED) 2] ls /brokers/ids
[0, 1, 2]

发现broker.id为0的这个节点信息也有了，在通过describe查看topic的描述信息，Isr中的信息和Replicas中的内容是一样的了

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic hello
Topic: hello    PartitionCount: 5       ReplicationFactor: 2    Configs: 
        Topic: hello    Partition: 0    Leader: 2       Replicas: 2,0       Isr: 2,0
        Topic: hello    Partition: 1    Leader: 1       Replicas: 0,1       Isr: 1,0
        Topic: hello    Partition: 2    Leader: 1       Replicas: 1,2       Isr: 1,2
        Topic: hello    Partition: 3    Leader: 2       Replicas: 2,1       Isr: 2,1
        Topic: hello    Partition: 4    Leader: 2       Replicas: 0,2       Isr: 2,0

但是启动后有个问题：发现新启动的这个节点不会是任何分区的leader？怎么重新均匀分配呢？

1、Broker中的自动均衡策略（默认已经有）

auto.leader.rebalance.enable=true
leader.imbalance.check.interval.seconds 默认值：300

2、手动执行：

bin/kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type preferred --all-topic-partitions

我们尝试使用手工执行

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type preferred --all-topic-partitions 
Successfully completed leader election (PREFERRED) for partitions hello-4, hello-1

执行后的效果如下，这样就实现了均匀分配

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic hello

Topic: hello    PartitionCount: 5       ReplicationFactor: 2    Configs: 
        Topic: hello    Partition: 0    Leader: 2       Replicas: 2,0Isr: 2,0
        Topic: hello    Partition: 1    Leader: 0       Replicas: 0,1Isr: 1,0
        Topic: hello    Partition: 2    Leader: 1       Replicas: 1,2Isr: 1,2
        Topic: hello    Partition: 3    Leader: 2       Replicas: 2,1Isr: 2,1
        Topic: hello    Partition: 4    Leader: 0       Replicas: 0,2Isr: 2,0

1.7 代码实战

前面我们使用基于console的生产者和消费者对topic实现了数据的生产和消费，，这个基于控制台的生产者和消费者主要是让我们做测试用的。在实际工作中，我们有时候需要将生产者和消费者功能集成到我们已有的系统中，此时就需要写代码实现生产者和消费者的逻辑了。在这我们使用java代码来实现生产者和消费者的功能

1.7.1 Kafka Java代码编程

先创建maven项目，db_kafka，并且添加kafka的maven依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.1</version>
</dependency>

开发生产者代码

package cn.git.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * 需求：Java代码实现生产者代码
 */
public class ProducerDemo {
   public static void main(String[] args) {
      Properties prop = new Properties();
      // 指定kafka的broker地址
      prop.put("bootstrap.servers", "bigdata01:9092,bigdata02:9092,bigdata03:9092");
      // 指定key-value数据的序列化格式
      prop.put("key.serializer", StringSerializer.class.getName());
      prop.put("value.serializer", StringSerializer.class.getName());

      // 指定topic
      String topic = "hello"; 
      
      // 创建kafka生产者
      KafkaProducer<String, String> producer = new KafkaProducer<String,String>(prop);

      // 向topic中生产数据
      producer.send(new ProducerRecord<String, String>(topic, "hello kafka"));

      // 关闭链接
      producer.close();
   }
}

开发消费者代码

package com.git.kafka;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
/**
 * 需求：Java代码实现消费者代码
 */
public class ConsumerDemo {
   public static void main(String[] args) {  
      Properties prop = new Properties();
      // 指定kafka的broker地址
      prop.put("bootstrap.servers", "bigdata01:9092,bigdata02:9092,bigdata03:9092");
      // 指定key-value的反序列化类型
      prop.put("key.deserializer", StringDeserializer.class.getName());
      prop.put("value.deserializer", StringDeserializer.class.getName());
      // 指定消费者组
      prop.put("group.id", "con-1");

      // 创建消费者
      KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(prop);
      Collection<String> topics = new ArrayList<String>();
      topics.add("hello");
      // 订阅指定的topic
      consumer.subscribe(topics);

      while(true) {
         // 消费数据【注意：需要修改jdk编译级别为1.8，否则Duration.ofSeconds(1)会语法报错】
         ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(1));
         for (ConsumerRecord<String,String> consumerRecord : poll) {
            System.out.println(consumerRecord);
         }
      }
   }
}

开发测试过程中注意事项：

1. 关闭kafka服务器的防火墙
2. 配置windows的hosts文件 添加kafka节点的hostname和ip的映射关系。如果我们的hosts文件中没有对kafka节点的 hostnam和ip的映射关系做配置，在这经过多次尝试连接不上就会报错

先开启消费者。发现没有消费到数据，这个topic中是有数据的，为什么之前的数据没有消费出来呢？不要着急，先带着这个问题往下面看

再开启生产者，生产者会生产一条数据，然后就结束

到消费者窗口看见消费数据

ConsumerRecord(topic = hello, partition = 0, leaderEpoch = 0, offset = 4, CreateTime = 1702437331857, serialized key size = -1, serialized value size = 20, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka 20231213)

所以这个时候我们发现，新产生的数据我们是可以消费到的，但是之前的数据我们就无法消费了，那下面我们来分析一下这个问题

1.7.2 消费者代码扩展

//==================================================
// 开启自动提交offset功能，默认就是开启的
prop.put("enable.auto.commit","true");
// 自动提交offset的时间间隔，单位是毫秒
prop.put("auto.commit.interval.ms","5000");
/*
注意：正常情况下，kafka消费数据的流程是这样的
先根据group.id指定的消费者组到kafka中查找之前保存的offset信息
如果查找到了，说明之前使用这个消费者组消费过数据，则根据之前保存的offset继续进行消费
如果没查找到(说明第一次消费)，或者查找到了，但是查找到的那个offset对应的数据已经不存在了
这个时候消费者该如何消费数据？
(因为kafka默认只会保存7天的数据，超过时间数据会被删除)

此时会根据auto.offset.reset的值执行不同的消费逻辑

这个参数的值有三种:[earliest,latest,none]
earliest：表示从最早的数据开始消费(从头消费)
latest【默认】：表示从最新的数据开始消费
none：如果根据指定的group.id没有找到之前消费的offset信息，就会抛异常

解释：【查找到了，但是查找到的那个offset对应的数据已经不存在了】
假设你第一天使用一个消费者去消费了一条数据，然后就把消费者停掉了，
等了7天之后，你又使用这个消费者去消费数据
这个时候，这个消费者启动的时候会到kafka里面查询它之前保存的offset信息
但是那个offset对应的数据已经被删了，所以此时再根据这个offset去消费是消费不到数据的


总结，一般在实时计算的场景下，这个参数的值建议设置为latest，消费最新的数据

这个参数只有在消费者第一次消费数据，或者之前保存的offset信息已过期的情况下才会生效
 */
prop.put("auto.offset.reset","latest");
//==================================================

此时我们来验证一下，先启动一次生产者,再启动一次消费者，看看消费者能不能消费到这条数据，如果能消费到，就说明此时是根据上次保存的offset信息进行消费了。结果发现是可以消费到的。

注意：消费者消费到数据之后，不要立刻关闭程序，要至少等5秒，因为自动提交offset的时机是5秒提交一次

ConsumerRecord(topic = hello, partition = 4, leaderEpoch = 5, offset = 0, CreateTime = 1591687894952, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)

将auto.offset.reset置为earliest，修改一下group.id的值，相当于使用一个新的消费者，验证一下，看是否能把这个topic中的所有数据都取出来，因为新的消费者第一次肯定是获取不到offset信息的，所以就会根据auto.offset.reset的值来消费数据

所以就会根据auto.offset.reset的值来消费数据

prop.put("group.id", "con-2");
prop.put("auto.offset.reset","earliest");

结果发现确实把之前的所有数据都消费过来了

ConsumerRecord(topic = hello, partition = 2, leaderEpoch = 0, offset = 0, CreateTime = 1591672800863, serialized key size = -1, serialized value size = 4, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hehe)
ConsumerRecord(topic = hello, partition = 3, leaderEpoch = 3, offset = 0, CreateTime = 1591687499753, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
ConsumerRecord(topic = hello, partition = 4, leaderEpoch = 5, offset = 0, CreateTime = 1591687864482, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)

此时，关闭消费者(需要等待5秒，这样才会提交offset)，再重新启动，发现没有消费到数据，说明此时就根据上次保存的offset来消费数据了，因为没有新数据产生，所以就消费不到了。

最后来处理一下程序输出的日志警告信息，这里其实示因为缺少依赖日志依赖，在pom文件中添加log4j的依赖，然后将log4j.properties添加到resources目录中

在pom文件中添加log4j的依赖，然后将log4j.properties添加到resources目录中

<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.10</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.10</version>
</dependency>

1.7.3 Consumer消费offset查询

kafka0.9版本以前，消费者的offset信息保存在zookeeper中，从kafka0.9开始，使用了新的消费API，消费者的信息会保存在kafka里面的__consumer_offsets这个topic中，因为频繁操作zookeeper性能不高，所以kafka在自己的topic中负责维护消费者的offset信息。

如何查询保存在kafka中的Consumer的offset信息呢？

使用kafka-consumer-groups.sh这个脚本可以查看，查看目前所有的consumer group

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-consumer-groups.sh --list  --bootstrap-server localhost:9092
con-1
con-2

具体查看某一个consumer group的信息

• GROUP：当前消费者组，通过group.id指定的值
• TOPIC：当前消费的topic
• PARTITION：消费的分区
• CURRENT-OFFSET：消费者消费到这个分区的offset
• LOG-END-OFFSET：当前分区中数据的最大offset
• LAG：当前分区未消费数据量

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-consumer-groups.sh --describe  --bootstrap-server localhost:9092 --group con-1
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
con-1           hello           4          1               1               0               -               -               -
con-1           hello           2          1               1               0               -               -               -
con-1           hello           3          1               1               0               -               -               -
con-1           hello           0          0               0               0               -               -               -
con-1           hello           1          0               0               0               -               -               -

此时再执行一次生产者代码，生产一条数据，重新查看一下这个消费者的offset情况

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-consumer-groups.sh --describe  --bootstrap-server localhost:9092 --group con-1
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
con-1           hello           4          1               2               1               -               -               -
con-1           hello           2          1               1               0               -               -               -
con-1           hello           3          1               1               0               -               -               -
con-1           hello           0          0               0               0               -               -               -
con-1           hello           1          0               0               0               -               -               -

1.7.4 Consumer消费顺序

当一个消费者消费一个partition时候，消费的数据顺序和此partition数据的生产顺序是一致的
当一个消费者消费多个partition时候，消费者按照partition的顺序，首先消费一个partition，当消费完一个partition最新的数据后再消费其它partition中的数据

总之：如果一个消费者消费多个partiton，只能保证消费的数据顺序在一个partition内是有序的

也就是说消费kafka中的数据只能保证消费partition内的数据是有序的，多个partition之间是无序的。

1.7.5 Kafka的三种语义

kafka可以实现以下三种语义，这三种语义是针对消费者而言的：

• 至少一次：at-least-once

  这种语义有可能会对数据重复处理，实现至少一次消费语义的消费者也很简单。
  1: 设置enable.auto.commit为false，禁用自动提交offset
  2: 消息处理完之后手动调用consumer.commitSync()提交offset

  这种方式是在消费数据之后，手动调用函数consumer.commitSync()异步提交offset，
  有可能处理多次的场景是消费者的消息处理完并输出到结果库，但是offset还没提交，这个时候消费者挂掉了，再重启的时候会重新消费并处理消息，所以至少会处理一次

• 至多一次：at-most-once，这种语义有可能会丢失数据

  至多一次消费语义是kafka消费者的默认实现。配置这种消费者最简单的方式是
  1: enable.auto.commit设置为true。
  2: auto.commit.interval.ms设置为一个较低的时间范围。

  由于上面的配置，此时kafka会有一个独立的线程负责按照指定间隔提交offset。

  消费者的offset已经提交，但是消息还在处理中(还没有处理完)，这个时候程序挂了，导致数据没有被成功处理，再重启的时候会从上次提交的offset处消费，导致上次没有被成功处理的消息就丢失了。

• 仅一次：exactly-once

  这种语义可以保证数据只被消费处理一次。实现仅一次语义的思路如下：
  1: 将enable.auto.commit设置为false，禁用自动提交offset
  2: 使用consumer.seek(topicPartition，offset)来指定offset
  3: 在处理消息的时候，要同时保存住每个消息的offset。以原子事务的方式保存offset和处理的消息结果，这个时候相当于自己保存offset信息了，把offset和具体的数据绑定到一块，数据真正处理成功的时候才会保存offset信息
  这样就可以保证数据仅被处理一次了。

1.8 Kafka集群参数调忧

1.8.1 JVM参数调忧

默认启动的Broker进程只会使用1G内存，在实际使用中会导致进程频繁GC，影响Kafka集群的性能和稳定性
通过jstat -gcutil <pid> 1000查看到kafka进程GC情况
主要看YGC,YGCT,FGC,FGCT这几个参数，如果这几个值不是很大，就没什么问题

YGC：young gc发生的次数
YGCT：young gc消耗的时间
FGC：full gc发生的次数
FGCT：full gc消耗的时间

[root@bigdata01 kafka_2.12-2.4.1]# jps
13248 Kafka
18087 Jps
1679 QuorumPeerMain
[root@bigdata01 kafka_2.12-2.4.1]# jstat -gcutil 13248 1000
 S0     S1     E      O      M     CCS    YGC     YGCT    FGC    FGCT     GCT   
  0.00 100.00  78.00  14.50  89.97  92.39     28    0.563     0    0.000    0.563
  0.00 100.00  78.00  14.50  89.97  92.39     28    0.563     0    0.000    0.563
  0.00 100.00  78.00  14.50  89.97  92.39     28    0.563     0    0.000    0.563
  0.00 100.00  78.00  14.50  89.97  92.39     28    0.563     0    0.000    0.563
  0.00 100.00  78.00  14.50  89.97  92.39     28    0.563     0    0.000    0.563
  0.00 100.00  78.00  14.50  89.97  92.39     28    0.563     0    0.000    0.563

如果你发现YGC很频繁，或者FGC很频繁，就说明内存分配的少了，此时需要修改kafka-server-start.sh中的 KAFKA_HEAP_OPTS

export KAFKA_HEAP_OPTS="-Xmx10g -Xms10g -XX:MetaspaceSize=96m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80"

这个配置表示给kafka分配了10G内存，我们的Kafka服务器是16G内存

1.8.2 Replication参数调忧

replica.socket.timeout.ms=60000：这个参数的默认值是30秒，它是控制partiton副本之间socket通信的超时时间，如果设置的太小，有可能会由于网络原因导致造成误判，认为某一个partition副本连不上了。

replica.lag.time.max.ms=50000：如果一个副本在指定的时间内没有向leader节点发送任何请求，或者在指定的时间内没有同步完leader中的数据，则leader会将这个节点从Isr列表中移除。这个参数的值默认为10秒，如果网络不好，或者kafka压力较大，建议调大该值，否则可能会频繁出现副本丢失，进而导致集群需要频繁复制副本，导致集群压力更大，会陷入一个恶性循环

1.8.3 Log参数调忧

这块是针对Kafka中数据文件的删除时机进行设置，不是对kafka本身的日志参数配置

log.retention.hours=24：这个参数默认值为168，单位是小时，就是7天，默认对数据保存7天，可以在这调整数据保存的时间，我们在实际工作中改为了只保存1天，因为kafka中的数据我们会在hdfs中进行备份，保存一份，所以就没有必要在kafka中保留太长时间了。

在kafka中保留只是为了能够让你在指定的时间内恢复数据，或者重新消费数据，如果没有这种需求，那就没有必要设置太长时间。这里分析的Replication的参数和Log参数都是在server.properties文件中进行配置，JVM参数是在kafka-server-start.sh脚本中配置

1.8.4 Kafka Topic命名小技巧

针对Kafka中Topic命名的小技巧，建议在给topic命名的时候在后面跟上r2p10之类的内容

• r2：表示Partition的副本因子是2
• p10：表示这个Topic的分区数是10

这样的好处是后期我们如果要写消费者消费指定topic的数据，通过topic的名称我们就知道应该设置多少个消费者消费数据效率最高。因为一个partition同时只能被一个消费者消费，所以效率最高的情况就是消费者的数量和topic的分区数量保持一致。在这里通过topic的名称就可以直接看到，一目了然。

但是也有一个缺点，就是后期如果我们动态调整了topic的partiton，那么这个topic名称上的partition数量就不准了，针对这个topic，建议大家一开始的时候就提前预估一下，可以多设置一些partition，我们在工作中的时候针对一些数据量比较大的topic一般会设置40~50个partition，数据量少的topic一般设置5~10个partition，这样后期调整topic partiton数量的场景就比较少了。

1.8.5 Kafka集群监控管理工具

现在我们操作Kafka都是在命令行界面中通过脚本操作的，后面需要传很多参数，用起来还是比较麻烦的，那kafka没有提供web界面的支持吗？很遗憾的告诉你，Apache官方并没有提供，不过好消息是有一个由雅虎开源的一个工具，目前用起来还是不错的。它之前的名字叫KafkaManager，后来改名字了，叫CMAK

CMAK是目前最受欢迎的Kafka集群管理工具，最早由雅虎开源，用户可以在Web界面上操作Kafka集群可以轻松检查集群状态(Topic、Consumer、Offset、Brokers、Replica、Partition)

那下面我们先去下载这个CMAK，需要到github上面去下载，在github里面搜索 CMAK即可

注意：由于cmak-3.0.0.4.zip版本是在java11这个版本下编译的，所以在运行的时候也需要使用java11这个版本，我们目前服务器上使用的是java8这个版本

我们为什么不使用java11版本呢？因为自2019年1月1日1起，java8之后的更新版本在商业用途的时候就需要收费授权了。

在这针对cmak-3.0.0.4这个版本，如果我们想要使用的话有两种解决办法
1：下载cmak的源码，使用jdk8编译
2：额外安装一个jdk11

如果想要编译的话需要安装sbt这个工具对源码进行编译， sbt 是 Scala 的构建工具, 类似于 Maven。由于我们在这使用不属于商业用途，所以使用jdk11是没有问题的，那就不用重新编译了。

下载jdk11，jdk-11.0.7_linux-x64_bin.tar.gz，将jdk11的安装包上传到bigdata01的/data/soft目录下，只需要解压即可，不需要配置环境变量，因为只有cmak这个工具才需要使用jdk11

[root@bigdata01 soft]# tar -zxvf jdk-11.0.7_linux-x64_bin.tar.gz

接下来把cmak-3.0.0.4.zip上传到bigdata01的/data/soft目录下

1：解压

[root@bigdata01 soft]# unzip cmak-3.0.0.4.zip  
-bash: unzip: command not found

注意：如果提示-bash: unzip: command not found，则说明目前不支持unzip命令，可以使用yum在线安装

建议先清空一下yum缓存，否则使用yum可能无法安装unzip

[root@bigdata01 soft]# yum clean all            
Loaded plugins: fastestmirror
Cleaning repos: base extras updates
Cleaning up list of fastest mirrors
34
[root@bigdata01 soft]# yum install -y unzip
Loaded plugins: fastestmirror
.....
Running transaction
  Installing : unzip-6.0-21.el7.x86_64                            1/1 
  Verifying  : unzip-6.0-21.el7.x86_64                            1/1 

Installed:
  unzip.x86_64 0:6.0-21.el7                                           

Complete!

再重新解压

[root@bigdata01 soft]# unzip cmak-3.0.0.4.zip

2：修改CMAK配置
首先修改bin目录下的cmak脚本，在里面配置JAVA_HOME指向jdk11的安装目录，否则默认会使用jdk8

[root@bigdata01 soft]# cd cmak-3.0.0.4
[root@bigdata01 cmak-3.0.0.4]# cd bin/
[root@bigdata01 bin]# vi cmak
....
JAVA_HOME=/data/soft/jdk-11.0.7
.....

然后修改conf目录下的application.conf文件，只需要在里面增加一行cmak.zkhosts参数的配置即可，指定zookeeper的地址

注意：在这里指定zookeeper地址主要是为了让CMAK在里面保存数据，这个zookeeper地址不一定是kafka集群使用的那个zookeeper集群，随便哪个zookeeper集群都可以。

[root@bigdata01 cmak-3.0.0.4]# cd conf/
[root@bigdata01 conf]# vi application.conf 
....
cmak.zkhosts="bigdata01:2181,bigdata02:2181,bigdata03:2181"
....

3：修改kafka启动配置
想要在CMAK中查看kafka的一些指标信息，在启动kafka的时候需要指定JMX_PORT,停止kafka集群

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-server-stop.sh 
[root@bigdata02 kafka_2.12-2.4.1]# bin/kafka-server-stop.sh 
[root@bigdata03 kafka_2.12-2.4.1]# bin/kafka-server-stop.sh 

重新启动kafka集群，指定JXM_PORT

[root@bigdata01 kafka_2.12-2.4.1]# JMX_PORT=9988 bin/kafka-server-start.sh -daemon config/server.properties
[root@bigdata02 kafka_2.12-2.4.1]# JMX_PORT=9988 bin/kafka-server-start.sh -daemon config/server.properties
[root@bigdata03 kafka_2.12-2.4.1]# JMX_PORT=9988 bin/kafka-server-start.sh -daemon config/server.properties

4：启动cmak

[root@bigdata01 cmak-3.0.0.4]# bin/cmak -Dconfig.file=conf/application.conf -Dhttp.port=9001

如果想把cmak放在后台执行的话需要添加上nohup和&

[root@bigdata01 cmak-3.0.0.4]# nohup bin/cmak -Dconfig.file=conf/application.conf -Dhttp.port=9001 &

5：访问 cmak http://bigdata01:9001/

6：操作CMAK

• 添加集群
  
  
  这几个参数配置好了以后还需要配置以下几个线程池相关的参数，这几个参数默认值是1，在保存的时候会提示需要大于1，所以可以都改为10

  brokerViewThreadPoolSize：10
  offsetCacheThreadPoolSize：10
  kafkaAdminClientThreadPoolSize：10
  

  最后点击Save按钮保存即可
  
  最后进来是这样的
  

• 查看kafak集群的所有broker信息
  

• 查看kafak集群的所有topic信息
  

• 查看某一个topic的详细信息
  
  

  点击topic的消费者信息是可以进来查看的

  

• 创建一个topic
  

• 给topic增加分区
  

  这是CMAK中常见的功能，当然了这里面还要一些我们没有说到的功能就留给大家以后来发掘了。

1.9 Flume集成Kafka

在实际工作中flume和kafka会深度结合使用

• 1：flume采集数据，将数据实时写入kafka
• 2：flume从kafka中消费数据，保存到hdfs，做数据备份

下面我们就来看一个综合案例，使用flume采集日志文件中产生的实时数据，写入到kafka中，然后再使用flume从kafka中将数据消费出来，保存到hdfs上面

那为什么不直接使用flume将采集到的日志数据保存到hdfs上面呢？因为中间使用kafka进行缓冲之后，后面既可以实现实时计算，又可以实现离线数据备份，最终实现离线计算，所以这一份数据就可以实现两种需求，使用起来很方便，所以在工作中一般都会这样做。

下面我们来实现一下这个功能，其实在Flume中，针对Kafka提供的有KafkaSource和KafkaSink

• KafkaSource是从kafka中读取数据
• KafkaSink是向kafka中写入数据

所以针对我们目前这个架构，主要就是配置Flume的Agent。需要配置两个Agent：

• 第一个Agent负责实时采集日志文件，将采集到的数据写入Kafka中

  针对第一个Agent：

  source：ExecSource，使用tail -F监控日志文件即可
  channel：MemoryChannel
  sink：KafkaSink

• 第二个Agent负责从Kafka中读取数据，将数据写入HDFS中进行备份

  针对第二个Agent
  Source：KafkaSource
  channel：MemoryChannel
  sink：HdfsSink

这里面这些组件其实只有KafkaSource和KafkaSink我们没有使用过，其它的组件都已经用过了。

下面来配置第一个Agent：文件名为：file-to-kafka.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /data/log/test.log

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Describe the sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
# 指定topic名称
a1.sinks.k1.kafka.topic = test_topic
# 指定kafka地址，多个节点地址使用逗号分割
# a1.sinks.k1.kafka.bootstrap.servers = bigdata01:9092,bigdata02:9092,bigdata03:9092
a1.sinks.k1.kafka.bootstrap.servers = bigdata01:9092

# 一次向kafka中写多少条数据，默认值为100，在这里为了演示方便，改为1
# 在实际工作中这个值具体设置多少需要在传输效率和数据延迟上进行取舍
# 如果kafka后面的实时计算程序对数据的要求是低延迟，那么这个值小一点比较好
# 如果kafka后面的实时计算程序对数据延迟没什么要求，那么就考虑传输性能，一次多传输一些数据，这样吞吐量会有所提升
# 建议这个值的大小和ExecSource每秒钟采集的数据量大致相等，这样不会频繁向kafka中写数据，并且对kafka后面的实时计算程序也没有很大影响，1秒的数据延迟一般是可以接收的
a1.sinks.k1.kafka.flumeBatchSize = 1
a1.sinks.k1.kafka.producer.acks = 1

# 一个Batch被创建之后，最多过多久，不管这个Batch有没有写满，都必须发送出去
# linger.ms和flumeBatchSize，哪个先满足先按哪个规则执行，这个值默认是0，在这设置为1表示每隔1毫秒就将这一个Batch中的数据发送出去
a1.sinks.k1.kafka.producer.linger.ms = 1
# 指定数据传输时的压缩格式，对数据进行压缩，提高传输效率
a1.sinks.k1.kafka.producer.compression.type = snappy

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

下面来配置第二个Agent：文件名为：kafka-to-hdfs.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
# 一次性向channel中写入的最大数据量，在这为了演示方便，设置为1
# 这个参数的值不要大于 MemoryChannel 中 transactionCapacity 的值
a1.sources.r1.batchSize = 1
# 最大多长时间向channel写一次数据
a1.sources.r1.batchDurationMillis = 2000
# kafka地址
# a1.sources.r1.kafka.bootstrap.servers = bigdata01:9092,bigdata02:9092,bigdata03:9092
a1.sources.r1.kafka.bootstrap.servers = bigdata01:9092
# topic名称，可以指定一个或者多个，多个topic之间使用逗号隔开
# 也可以使用正则表达式指定一个topic名称规则
a1.sources.r1.kafka.topics = test_topic
# 指定消费者组id
a1.sources.r1.kafka.consumer.group.id = flume-con1

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://bigdata01:9000/kafkaout
a1.sinks.k1.hdfs.filePrefix = data-
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

在bigdata04机器的flume目录下复制两个目录

[root@bigdata04 apache-flume-1.9.0-bin]# cd /data/soft/apache-flume-1.9.0-bin
[root@bigdata04 apache-flume-1.9.0-bin]# cp -r conf conf-file-to-kafka
[root@bigdata04 apache-flume-1.9.0-bin]# cp -r conf conf-kafka-to-hdfs

修改 conf_file_to_kafka和conf_kafka_to_hdfs中log4j的配置

[root@bigdata04 apache-flume-1.9.0-bin]# cd conf-file-to-kafka
[root@bigdata04 conf_file_to_kafka]# vi log4j.properties 
flume.root.logger=ERROR,LOGFILE
flume.log.file=flume-file-to-kafka.log
==============================================================
[root@bigdata04 apache-flume-1.9.0-bin]# cd conf-kafka-to-hdfs
[root@bigdata04 conf_kafka_to_hdfs]# vi log4j.properties 
flume.root.logger=ERROR,LOGFILE
flume.log.file=flume-kafka-to-hdfs.log

把刚才配置的两个Agent的配置文件复制到这两个目录下

[root@bigdata04 apache-flume-1.9.0-bin]# cd conf-file-to-kafka
[root@bigdata04 conf-file-to-kafka]# vi file-to-kafka.conf
.....把file-to-kafka.conf文件中的内容复制进来即可
===============================================================
[root@bigdata04 apache-flume-1.9.0-bin]# cd conf-kafka-to-hdfs/
[root@bigdata04 conf-kafka-to-hdfs]# vi kafka-to-hdfs.conf
.....把kafka-to-hdfs.conf文件中的内容复制进来即可

启动这两个Flume Agent，确保zookeeper集群、kafka集群和Hadoop集群是正常运行的，以及Kafka中的topic需要提前创建好

创建topic

[root@bigdata01 kafka_2.12-2.4.1]# bin/kafka-topics.sh --create --zookeeper localhost:2181  --partitions 5 --replication-factor 2 --topic test_topic

先启动第二个Agent，再启动第一个Agent

[root@bigdata04 apache-flume-1.9.0-bin]# bin/flume-ng agent --name a1 --conf conf-kafka-to-hdfs --conf-file conf-kafka-to-hdfs/kafka-to-hdfs.conf

[root@bigdata04 apache-flume-1.9.0-bin]# bin/flume-ng agent --name a1 --conf conf-file-to-kafka --conf-file conf-file-to-kafka/file-to-kafka.conf

模拟产生日志数据

[root@bigdata04 ~]# cd /data/log/
[root@bigdata04 log]# echo hello world >> /data/log/test.log

到HDFS上查看数据，验证结果：

[root@bigdata04 ~]# hdfs dfs -ls /kafkaout
Found 1 items
-rw-r--r--   2 root supergroup         12 2020-06-09 22:59 /kafkaout/data-.1591714755267.tmp
[root@bigdata04 ~]# hdfs dfs -cat /kafkaout/data-.1591714755267.tmp
hello world

此时Flume可以通过tail -F命令实时监控文件中的新增数据，发现有新数据就写入kafka，然后kafka后面的flume落盘程序，以及kafka后面的实时计算程序就可以使用这份数据了。

2.0 Kafka集群平滑升级

之前我们在使用Kafka 0.9.0.0版本的时候，遇到一个比较诡异的问题
针对消费者组增加消费者的时候可能会导致rebalance，进而导致部分consumer不能再消费分区数据

意思就是之前针对这个topic的5个分区只有2个消费者消费数据，后期我动态的把消费者调整为了5个，这样可能会导致部分消费者无法消费分区中的数据。

针对这个bug这里有一份详细描述：
https://issues.apache.org/jira/browse/KAFKA-2978

此bug官方在0.9.0.1版本中进行了修复
当时我们的线上集群使用的就是0.9.0.0的版本。
所以我们需要对线上集群在不影响线上业务的情况下进行升级，称为平滑升级，也就是升级的时候不影响线上的正常业务运行。

接下来我们就查看了官网文档，上面有针对集群平滑升级的一些信息 http://kafka.apache.org/090/documentation.html#upgrade

在验证这个升级流程的时候我们是在测试环境下，先模拟线上的集群环境，进行充分测试，可千万不能简单测试一下就直接搞到测试环境去做，这样是很危险的。

由于当时这个kafka集群我们还没有移交给运维负责，并且运维当时对这个框架也不是很熟悉，所以才由我们开发人员来进行平滑升级，否则这种框架升级的事情肯定是交给运维去做的。

那接下来看一下具体的平滑升级步骤

小版本之间集群升级不需要额外修改集群的配置文件。只需要按照下面步骤去执行即可。
假设kafka0.9.0.0集群在三台服务器上，需要把这三台服务器上的kafka集群升级到0.9.0.1版本

注意：提前在集群的三台机器上把0.9.0.1的安装包，解压、配置好。

主要是log.dirs这个参数，0.9.0.1中的这个参数和0.9.0.0的这个参数一定要保持一致，这样新版本的kafka才可以识别之前的kakfa中的数据。
在集群升级的过程当中建议通过CMAK(kafkamanager)查看集群的状态信息，比较方便

1：先stop掉0.9.0.0集群中的第一个节点，然后去CMAK上查看集群的broker信息，确认节点确实已停掉。并且再查看一下，节点的副本下线状态。确认集群是否识别到副本下线状态。

然后在当前节点把kafka0.9.0.1启动起来。再回到CMAK中查看broker信息，确认刚启动的节点是否已正确显示，并且还要确认这个节点是否可以正常接收和发送数据。

2：按照第一步的流程去依次操作剩余节点即可，就是先把0.9.0.0版本的kafka停掉，再把0.9.0.1版本的kafka启动即可。

注意：每操作一个节点，需要稍等一下，确认这个节点可以正常接收和发送数据之后，再处理下一个节点。