【Kafka】Apache Kafka消息队列概述及环境搭建

冷不防 2023-10-05 12:23 98阅读 0赞

#### 指路牌 ####

*   *  概述
     *  主流MQ的比较
     *  架构
     *   *  日志&分区
         *  生产者
         *  消费组
         *  顺序写入&ZeroCopy
     *  Kafka环境搭建
     *   *  单机环境
         *  集群环境

### 概述 ###

**Apache Kafka**是Apache软件基金会的开源的**流处理平台**，该平台提供了消息的**订阅**与**发布**的消息队列，一般用作系统间 *解耦、异步通信、削峰填谷* 等场景。

同时Kafka又提供了**Kafka streaming**插件包实现了实时在线流处理。相比较一些专业的流处理框架不同，Kafka Streaming计算是运行在应用端，具有简单、入门要求低、部署方便等优点。总体来说Kafka这门课程需要大家掌握以下两个方面：

*  消息队列(MQ) Message Queue
 *  Kafka Streaming 流处理  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODIzMTQ0OA_size_16_color_FFFFFF_t_70]

### 主流MQ的比较 ###

目前市面上常见的MQ大概有以下几个：

*  **ActiveMQ：**  
    Apache下的一个子项目。使用Java完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，少量代码就可以高效地实现高级应用场景。可插拔的传输协议支持，比如：in-VM, TCP, SSL, NIO, UDP, multicast, JGroups and JXTA transports。RabbitMQ、ZeroMQ、ActiveMQ均支持常用的多种语言客户端 C++、Java、.Net,、Python、 Php、 Ruby等。
 *  **RocketMQ：**  
    阿里系下开源的一款分布式、队列模型的消息中间件，原名Metaq，3.0版本名称改为RocketMQ，是阿里参照kafka设计思想使用java实现的一套mq。同时将阿里系内部多款mq产品（Notify、metaq）进行整合，只维护核心功能，去除了所有其他运行时依赖，保证核心功能最简化，在此基础上配合阿里上述其他开源产品实现不同场景下mq的架构，目前主要多用于订单交易系统。
    
    *具有以下特点：*  
    能够保证严格的消息顺序  
    提供针对消息的过滤功能  
    提供丰富的消息拉取模式  
    高效的订阅者水平扩展能力  
    实时的消息订阅机制  
    亿级消息堆积能力
 *  **RabbitMQ：**  
    使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP,STOMP，也正是如此，使的它变的非常重量级，更适合于企业级的开发。同时实现了Broker架构，核心思想是生产者不会将消息直接发送给队列，消息在发送给客户端时先在中心队列排队。对路由(Routing)，负载均衡(Load balance)、数据持久化都有很好的支持。多用于进行企业级的ESB整合。
 *  **Redis：**  
    使用C语言开发的一个Key-Value的NoSQL数据库，开发维护很活跃，虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。
 *  **Kafka：**  
    Apache下的一个子项目，使用scala实现的一个高性能分布式Publish/Subscribe消息队列系统，具有以下特性：
    
     *  快速持久化：通过磁盘顺序读写与零拷贝机制，可以在O(1)的系统开销下进行消息持久化；
     *  高吞吐：在一台普通的服务器上既可以达到10W/s的吞吐速率；
     *  高堆积：支持topic下消费者较长时间离线，消息堆积量大；
     *  完全的分布式系统：Broker、Producer、Consumer都原生自动支持分布式，依赖zookeeper自动实现复杂均衡；
     *  支持Hadoop数据并行加载：对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。

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

消息队列是一种在分布式和大数据开发中不可或缺的中间件。在分布式或者大数据开发中通常使用消息队列进行**缓冲**、**系统间解耦**和**削峰填谷**等业务场景，常见的消息队列服务工作模式大致会分为两大类：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODIzMTQ0OA_size_16_color_FFFFFF_t_70 1]

*  至多一次：消息生产者将数据写入消息系统，然后由消费者负责去拉去消息服务器中的消息，一旦消息被确认消费之后 ，由消息服务器主动删除队列中的数据，这种消费方式一般只允许被一个消费者消费，并且消息队列中的数据不允许被重复消费。
 *  没有限制：同上述消费形式不同，生产者发不完数据以后，该消息可以被多个消费者同时消费，并且同一个消费者可以多次消费消息服务器中的同一个记录。主要是因为消息服务器一般可以长时间存储海量消息。

### 架构 ###

#### 日志&分区 ####

Kafka集群以**Topic**形式负责分类集群中的**Record**，每一个Record属于一个Topic，生产者负责发送数据到Kafka集群中的某一个Topic中，同时消费者可以订阅kafka集群中的Topic.  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sScYm287-1603547057588)(assets/20200104182508649.png)\]][img-sScYm287-1603547057588_assets_20200104182508649.png]

每个Topic底层都会对应一组分区的日志用于持久化Topic中的Record。在Kafka集群中，Topic的每一个日志的分区都一定会有1个Borker担当该分区的Leader，其他的Broker担当该分区的follower，Leader负责分区数据的读写操作，follower负责同步改分区的数据。这样如果分区的Leader宕机，改分区的其他follower会选取出新的leader继续负责该分区数据的读写。其中集群的中Leader的监控和Topic的部分元数据是存储在Zookeeper中。  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-h17kZwGD-1603547057590)(assets/20200104182529801.png)\]][img-h17kZwGD-1603547057590_assets_20200104182529801.png]

#### 生产者 ####

生产者将数据发布到他们选择的Topic。生产者负责选择将记录分配给Topic中的哪个Partition。可以以round-robin方式完成此操作，仅是为了平衡负载，也可以根据某些语义分区功能（例如基于记录中的Key）进行此操作

Kafka中所有消息是通过Topic为单位进行管理，每个Kafka中的Topic通常会有多个订阅者，负责订阅发送到改Topic中的数据。每个在Kafka 集群中的Topic，Kafka负责管理改Topic的一组日志分区：  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aKbGYdJW-1603547057593)(assets/20200104182545392.png)\]][img-aKbGYdJW-1603547057593_assets_20200104182545392.png]

每组日志分区是一个有序的不可变的的日志序列，分区中的每一个Record都被分配了唯一的序列编号称为是offset，Kafka 集群会持久化所有发布到Topic中的Record信息，改Record的持久化时间是通过配置文件指定,默认是168小时。

log.retention.hours=168

Kafka底层会定期的check日志文件，然后将过期的数据从log中移除，由于Kafka使用硬盘存储日志文件，因此使用Kafka长时间缓存一些日志文件是不存在问题的。

#### 消费组 ####

在消费者消费Topic中数据的时候，每个消费者会维护本次消费对应分区的偏移量，消费者会在消费完一个批次的数据之后，会将本次消费的偏移量提交给Kafka集群，因此对于每个消费者而言可以随意的控制该消费者的偏移量。因此在Kafka中，消费者可以从一个topic分区中的任意位置读取队列数据，由于每个消费者控制了自己的消费的偏移量，因此多个消费者之间彼此相互独立。消费者使用ConsumerGroup名称标记自己，并且发布到Topic的每条记录都会传递到每个订阅ConsumerGroup中的一个消费者实例。如果所有Consumer实例都具有相同的ConsumerGroup，那么Topic中的记录会在该ConsumerGroup中的Consumer实例进行均分消费；如果所有Consumer实例具有不同的ConsumerGroup，则每条记录将广播到所有ConsumerGroup进程。

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Df1WeqBg-1603547057599)(assets/20200104182620255.png)\]][img-Df1WeqBg-1603547057599_assets_20200104182620255.png]

更常见的是，我们发现Topic具有少量的Consumer Group，每个ConsumerGroup可以理解为一个“逻辑的订阅者”。每个ConsumerGroup均由许多Consumer实例组成，以实现可伸缩性和容错能力。这无非就是发布-订阅模型，其中订阅者是消费者的集群而不是单个进程。这种消费方式Kafka会将Topic按照分区的方式均分给一个ConsumerGroup下的实例，如果ConsumerGroup下有新的成员介入，则新介入的Consumer实例会去接管ConsumerGroup内其他消费者负责的某些分区，同样如果一下ConsumerGroup下的有其他Consumer实例宕机，则由该ConsumerGroup其他实例接管。

> 由于Kafka的Topic的分区策略，因此Kafka仅提供分区中记录的有序性，也就意味着相同Topic的不同分区记录之间无顺序。因为针对于绝大多数的大数据应用和使用场景， 使用分区内部有序或者使用key进行分区策略已经足够满足绝大多数应用场景。但是，如果您需要记录全局有序，则可以通过只有一个分区Topic来实现，尽管这将意味着每个ConsumerGroup只有一个Consumer进程。

#### 顺序写入&ZeroCopy ####

Kafka的特性之一就是高吞吐率，但是Kafka的消息是保存或缓存在磁盘上的，一般认为在磁盘上读写数据是会降低性能的，但是Kafka即使是普通的服务器，Kafka也可以轻松支持每秒百万级的写入请求，超过了大部分的消息中间件，这种特性也使得Kafka在日志处理等海量数据场景广泛应用。Kafka会把收到的消息都写入到硬盘中，防止丢失数据。为了优化写入速度Kafka采用了两个技术， 顺序写和MMFile 。

因为硬盘是机械结构，每次读写都会寻址，然后写入，其中寻址是一个“机械动作”，它是最耗时的。所以硬盘最讨厌随机I/O，最喜欢顺序I/O。为了提高读写硬盘的速度，Kafka就是使用顺序I/O。这样省去了大量的内存开销以及节省了IO寻址的时间。但是单纯的使用顺序写入，Kafka的写入性能也不可能和内存进行对比，因此Kafka的数据并不是实时的写入硬盘中 ，

Kafka充分利用了现代操作系统分页存储来利用内存提高I/O效率。Memory Mapped Files(后面简称mmap)也称为内存映射文件，在64位操作系统中一般可以表示20G的数据文件，它的工作原理是直接利用操作系统的Page来实现文件到物理内存的直接映射。完成MMP映射后，用户对内存的所有操作会被操作系统自动的刷新到磁盘上，极大地降低了IO使用率。  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-CLJAGOSH-1603547057602)(assets/20200104182633932.png)\]][img-CLJAGOSH-1603547057602_assets_20200104182633932.png]

Kafka服务器在响应客户端读取的时候，底层使用ZeroCopy技术，直接将磁盘无需拷贝到用户空间，而是直接将数据通过内核空间传递输出，数据并没有抵达用户空间。常规的IO操作流程如下：

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-m9UTim2q-1603547057607)(assets/20200104182654616.png)\]][img-m9UTim2q-1603547057607_assets_20200104182654616.png]

1.  用户进程调用read等系统调用向操作系统发出IO请求，请求读取数据到自己的内存缓冲区中。自己进入阻塞状态
2.  操作系统收到请求后，进一步将IO请求发送磁盘
3.  磁盘驱动器收到内核的IO请求，把数据从磁盘读取到驱动器的缓冲中。此时不占用CPU。当驱动器的缓冲区被读满后，向内核发起中断信号告知自己缓冲区已满
4.  内核收到中断，使用CPU时间将磁盘驱动器的缓存中的数据拷贝到内核缓冲区中。
5.  如果内核缓冲区的数据少于用户申请的读的数据，重复步骤3跟步骤4，直到内核缓冲区的数据足够多为止。
6.  将数据从内核缓冲区拷贝到用户缓冲区，同时从系统调用中返回完成读取任务

缺点：用户的每次IO请求，都需要CPU多次参与。

现代操作系统大都引入了协处理器的概念，也就是说系统在读取的磁盘文件的时候，无需CPU直接参与。而是将数据读取的任务交给DMA控制器，负责辅助CPU完成数据的读取过程。  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VBp7r8QY-1603547057608)(assets/20200104182707774.png)\]][img-VBp7r8QY-1603547057608_assets_20200104182707774.png]

1.  用户进程调用read等系统调用向操作系统发出IO请求，请求读取数据到自己的内存缓冲区中。自己进入阻塞状态。
2.  操作系统收到请求后，进一步将IO请求发送DMA。然后让CPU干别的活去。
3.  DMA进一步将IO请求发送给磁盘。
4.  磁盘驱动器收到DMA的IO请求，把数据从磁盘读取到驱动器的缓冲中。当驱动器的缓冲区被读满后，向DMA发起中断信号告知自己缓冲区已满。
5.  DMA收到磁盘驱动器的信号，将磁盘驱动器的缓存中的数据拷贝到内核缓冲区中。此时不占用CPU。这个时候只要内核缓冲区的数据少于用户申请的读的数据，内核就会一直重复步骤3跟步骤4，直到内核缓冲区的数据足够多为止。
6.  当DMA读取了足够多的数据，就会发送中断信号给CPU。
7.  CPU手动DMA的信号，知道数据已经准备好，于是将数据从内核拷贝到用户空间，系统调用返回。  
    跟IO中断模式相比，DMA模式下，DMA就是CPU的一个代理，它负责了一部分的拷贝工作，从而减轻了CPU的负担。DMA的优点就是：中断少，CPU负担低。

在了解底层操作系统的IO操作以后，我们来看一下网络场景。文件在磁盘中数据被copy到内核缓冲区 ->从内核缓冲区copy到用户缓冲区->用户缓冲区copy到内核与socket相关的缓冲区-> 数据从socket缓冲区copy到相关协议引擎发送出去。

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vo93m1w5-1603547057610)(assets/20200104182725874.png)\]][img-vo93m1w5-1603547057610_assets_20200104182725874.png]

从上图可以看出，默认数据需要从内核到用户在到内核空间的这么一个过程是多余的，所谓ZeroCopy的做法是直接在内核中将数据发送出去，因此节省了1次拷贝。文件在磁盘中数据被copy到内核缓冲区->从内核缓冲区copy到内核与socket相关的缓冲区->数据从socket缓冲区copy到相关协议引擎发送出去。

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-UyoNwYZW-1603547057613)(assets/20200104182753258.png)\]][img-UyoNwYZW-1603547057613_assets_20200104182753258.png]

### Kafka环境搭建 ###

#### 单机环境 ####

*  安装JDK，配置JAVA\_HOME

[root@CentOS ~]# rpm -ivh jdk-8u191-linux-x64.rpm 
    warning: jdk-8u191-linux-x64.rpm: Header V3 RSA/SHA256 Signature, key ID ec551f03: NOKEY
    Preparing...                ########################################### [100%]
       1:jdk1.8                 ########################################### [100%]
    Unpacking JAR files...
    	tools.jar...
    	plugin.jar...
    	javaws.jar...
    	deploy.jar...
    	rt.jar...
    	jsse.jar...
    	charsets.jar...
    	localedata.jar...
    [root@CentOS ~]# vi .bashrc 
    JAVA_HOME=/usr/java/latest
    CLASSPATH=.
    PATH=$PATH:$JAVA_HOME/bin
    export JAVA_HOME
    export CLASSPATH
    export PATH
    [root@CentOS ~]# vi .bashrc

*  配置主机名为CentOS

[root@CentOS ~]# vi /etc/sysconfig/network
    NETWORKING=yes
    HOSTNAME=CentOS

*  配置主机名和IP映射

[root@CentOS ~]# ifconfig 
    eth0      Link encap:Ethernet  HWaddr 00:0C:29:D3:EA:13  
              inet addr:192.168.52.129  Bcast:192.168.52.255  Mask:255.255.255.0
              inet6 addr: fe80::20c:29ff:fed3:ea13/64 Scope:Link
              UP BROADCAST RUNNING MULTICAST  MTU:1500  Metric:1
              RX packets:641 errors:0 dropped:0 overruns:0 frame:0
              TX packets:379 errors:0 dropped:0 overruns:0 carrier:0
              collisions:0 txqueuelen:1000 
              RX bytes:56374 (55.0 KiB)  TX bytes:57374 (56.0 KiB)
    
    lo        Link encap:Local Loopback  
              inet addr:127.0.0.1  Mask:255.0.0.0
              inet6 addr: ::1/128 Scope:Host
              UP LOOPBACK RUNNING  MTU:65536  Metric:1
              RX packets:0 errors:0 dropped:0 overruns:0 frame:0
              TX packets:0 errors:0 dropped:0 overruns:0 carrier:0
              collisions:0 txqueuelen:0 
              RX bytes:0 (0.0 b)  TX bytes:0 (0.0 b)
    [root@CentOS ~]# vi /etc/hosts      
    127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
    ::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
    192.168.52.129 CentOS

*  关闭防火墙

[root@CentOS ~]# service iptables stop
    iptables: Setting chains to policy ACCEPT: filter          [  OK  ]
    iptables: Flushing firewall rules:                         [  OK  ]
    iptables: Unloading modules:                               [  OK  ]
    [root@CentOS ~]# chkconfig iptables off
    [root@CentOS ~]# chkconfig --list | grep iptables
    iptables       	0:关闭	1:关闭	2:关闭	3:关闭	4:关闭	5:关闭	6:关闭

*  安装配置Zookeeper

[root@CentOS ~]# tar -zxf zookeeper-3.4.6.tar.gz -C /usr/
    [root@CentOS ~]# cd /usr/zookeeper-3.4.6/
    [root@CentOS zookeeper-3.4.6]# cp conf/zoo_sample.cfg conf/zoo.cfg
    [root@CentOS zookeeper-3.4.6]# vi conf/zoo.cfg 
    # The number of milliseconds of each tick
    tickTime=2000
    # The number of ticks that the initial
    # synchronization phase can take
    initLimit=10
    # The number of ticks that can pass between
    # sending a request and getting an acknowledgement
    syncLimit=5
    # the directory where the snapshot is stored.
    # do not use /tmp for storage, /tmp here is just
    # example sakes.
    dataDir=/root/zkdata
    # the port at which the clients will connect
    clientPort=2181
    # the maximum number of client connections.
    # increase this if you need to handle more clients
    #maxClientCnxns=60
    #
    # Be sure to read the maintenance section of the
    # administrator guide before turning on autopurge.
    #
    # http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
    #
    [root@CentOS ~]# mkdir /root/zkdata

*  启动zookeeper服务

[root@CentOS zookeeper-3.4.6]# ./bin/zkServer.sh 
    JMX enabled by default
    Using config: /usr/zookeeper-3.4.6/bin/../conf/zoo.cfg
    Usage: ./bin/zkServer.sh {
        start|start-foreground|stop|restart|status|upgrade|print-cmd}
    [root@CentOS zookeeper-3.4.6]# ./bin/zkServer.sh start zoo.cfg
    JMX enabled by default
    Using config: /usr/zookeeper-3.4.6/bin/../conf/zoo.cfg
    Starting zookeeper ... STARTED
    [root@CentOS zookeeper-3.4.6]# 
    [root@CentOS zookeeper-3.4.6]# ./bin/zkServer.sh status zoo.cfg
    JMX enabled by default
    Using config: /usr/zookeeper-3.4.6/bin/../conf/zoo.cfg
    Mode: standalone
    [root@CentOS zookeeper-3.4.6]# jps
    1778 Jps
    1733 QuorumPeerMain

*  安装配置Kafka集群

[root@CentOS ~]# tar -zxf kafka_2.11-2.2.0.tgz -C /usr/
    [root@CentOS ~]# cd /usr/kafka_2.11-2.2.0/
    [root@CentOS kafka_2.11-2.2.0]# vi config/server.properties 
    ############################# Server Basics #############################
    
    # The id of the broker. This must be set to a unique integer for each broker.
    broker.id=0
    
    ############################# Socket Server Settings #############################
    
    listeners=PLAINTEXT://CentOS:9092
    
    ############################# Log Basics #############################
    
    log.dirs=/usr/kafka-logs
    
    ############################# Log Retention Policy #############################
    
    log.retention.hours=168
    
    ############################# Zookeeper #############################
    
    zookeeper.connect=CentOS:2181
    
    [root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-server-start.sh -daemon config/server.properties

*  测试Kafka服务
    
    *创建Topic*

[root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-topics.sh \
                                                    --create 
                                                    --bootstrap-server CentOS:9092  \
                                                    --topic topic01  \
                                                    --partitions 3 \
                                                    --replication-factor 1
                                                    
    [root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-topics.sh \
                                                    --create \
                                                    --zookeeper CentOS:2181  \
                                                    --topic topic02  \
                                                    --partitions 3 \
                                                    --replication-factor 1

> kafka自2.2.0版本以后,Toipic的管理使用的的是`--bootstrap-server`不在使用`--zookeeper`，`--partitions`:指定分区数、`--replication-factor`指定副本因子数，该副本因子不能大于可用的broker节点的个数

*查看Topic列表*

[root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-topics.sh \
                                                    --bootstrap-server CentOS:9092 \
                                                    --list 
    topic01
    topic02

*查看Topic详情*

[root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-topics.sh \
                                    --describe \
                                    --bootstrap-server CentOS:9092  \
                                    --topic topic01
    Topic:topic01   PartitionCount:3        ReplicationFactor:1     Configs:segment.bytes=1073741824
            Topic: topic01  Partition: 0    Leader: 0       Replicas: 0     Isr: 0
            Topic: topic01  Partition: 1    Leader: 0       Replicas: 0     Isr: 0
            Topic: topic01  Partition: 2    Leader: 0       Replicas: 0     Isr: 0

> `segment.bytes`:Kafka底层在存储分区的文件的时候是按照段落存储的，也就是某个分区的文件达到1GB（1073741824 bytes）的时候，系统会生成新的段落，这种设计有助于Broker节点索引文件。`Replicas`:表示副本集成员broker-id，`Isr`:表示处于同步中的正常副本集全称(In Synch Replicate)

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-yywvsD8C-1603547057616)(assets/image-20201019120149718.png)\]][img-yywvsD8C-1603547057616_assets_image-20201019120149718.png]

订阅topic

[root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-console-consumer.sh \
                                                  --bootstrap-server CentOS:9092 \
                                                  --group g1  \
                                                  --topic topic01

*查看消费组*

[root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-consumer-groups.sh \
    															--bootstrap-server CentOS:9092  \
    															--describe \
    															--group g1
    
    TOPIC  PARTITION CURRENT-OFFSET LOG-END-OFFSET  LAG  CONSUMER-ID  HOST            CLIENT-ID
    topic01 0          0               0             0   consumer-** /192.168.52.129  consumer-1
    
    [root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-consumer-groups.sh 
                                  --bootstrap-server CentOS:9092  
                                  --describe 
                                  --group g1 
                                  --members
    
    CONSUMER-ID                  HOST          CLIENT-ID       #PARTITIONS     
    consumer-***              /192.168.52.129  consumer-1      1

生产消息

[root@CentOS kafka_2.11-2.2.0]# ./bin/kafka-console-producer.sh \
                                    --broker-list CentOS:9092  \
                                    --topic topic01
    >

#### 集群环境 ####

*  安装JDK，配置JAVA\_HOME

[root@CentOSX ~]# rpm -ivh jdk-8u191-linux-x64.rpm 
    warning: jdk-8u191-linux-x64.rpm: Header V3 RSA/SHA256 Signature, key ID ec551f03: NOKEY
    Preparing...                ########################################### [100%]
       1:jdk1.8                 ########################################### [100%]
    Unpacking JAR files...
    	tools.jar...
    	plugin.jar...
    	javaws.jar...
    	deploy.jar...
    	rt.jar...
    	jsse.jar...
    	charsets.jar...
    	localedata.jar...
    [root@CentOSX ~]# vi .bashrc 
    JAVA_HOME=/usr/java/latest
    CLASSPATH=.
    PATH=$PATH:$JAVA_HOME/bin
    export JAVA_HOME
    export CLASSPATH
    export PATH
    [root@CentOSX ~]# vi .bashrc

*  配置主机名为CentOS

[root@CentOS ~]# vi /etc/sysconfig/network
    NETWORKING=yes
    HOSTNAME=CentOS[A,B,C]

*  配置主机名和IP映射

[root@CentOS ~]# vi /etc/hosts      
    127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
    192.168.52.130 CentOSA
    192.168.52.131 CentOSB
    192.168.52.132 CentOSC

*  关闭防火墙

[root@CentOSX ~]# service iptables stop
    iptables: Setting chains to policy ACCEPT: filter          [  OK  ]
    iptables: Flushing firewall rules:                         [  OK  ]
    iptables: Unloading modules:                               [  OK  ]
    [root@CentOSX ~]# chkconfig iptables off
    [root@CentOSX ~]# chkconfig --list | grep iptables
    iptables       	0:关闭	1:关闭	2:关闭	3:关闭	4:关闭	5:关闭	6:关闭

*  安装配置Zookeeper

[root@CentOSX ~]# tar -zxf zookeeper-3.4.6.tar.gz -C /usr/
    [root@CentOSX ~]# cd /usr/zookeeper-3.4.6/
    [root@CentOSX zookeeper-3.4.6]# cp conf/zoo_sample.cfg conf/zoo.cfg
    [root@CentOSX zookeeper-3.4.6]# vi conf/zoo.cfg 
    # The number of milliseconds of each tick
    tickTime=2000
    # The number of ticks that the initial
    # synchronization phase can take
    initLimit=10
    # The number of ticks that can pass between
    # sending a request and getting an acknowledgement
    syncLimit=5
    # the directory where the snapshot is stored.
    # do not use /tmp for storage, /tmp here is just
    # example sakes.
    dataDir=/root/zkdata
    # the port at which the clients will connect
    clientPort=2181
    # the maximum number of client connections.
    # increase this if you need to handle more clients
    #maxClientCnxns=60
    #
    # Be sure to read the maintenance section of the
    # administrator guide before turning on autopurge.
    #
    # http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
    #
    server.1=CentOSA::2888:3888
    server.2=CentOSA::2888:3888
    server.3=CentOSA::2888:3888
    [root@CentOSX ~]# mkdir /root/zkdata
    [root@CentOSA ~]# echo 1 > /root/zkdata/myid
    [root@CentOSB ~]# echo 2 > /root/zkdata/myid
    [root@CentOSC ~]# echo 3 > /root/zkdata/myid

*  启动zookeeper服务

[root@CentOSX zookeeper-3.4.6]# ./bin/zkServer.sh 
    JMX enabled by default
    Using config: /usr/zookeeper-3.4.6/bin/../conf/zoo.cfg
    Usage: ./bin/zkServer.sh {
        start|start-foreground|stop|restart|status|upgrade|print-cmd}
    [root@CentOS zookeeper-3.4.6]# ./bin/zkServer.sh start zoo.cfg
    JMX enabled by default
    Using config: /usr/zookeeper-3.4.6/bin/../conf/zoo.cfg
    Starting zookeeper ... STARTED
    [root@CentOS zookeeper-3.4.6]# 
    [root@CentOS zookeeper-3.4.6]# ./bin/zkServer.sh status zoo.cfg
    JMX enabled by default
    Using config: /usr/zookeeper-3.4.6/bin/../conf/zoo.cfg
    Mode: leader|follower
    [root@CentOSX zookeeper-3.4.6]# jps
    1778 Jps
    1733 QuorumPeerMain

*  安装配置Kafka集群

[root@CentOSX ~]# tar -zxf kafka_2.11-2.2.0.tgz -C /usr/
    [root@CentOSX ~]# cd /usr/kafka_2.11-2.2.0/
    [root@CentOSX kafka_2.11-2.2.0]# vi config/server.properties 
    ############################# Server Basics #############################
    
    # The id of the broker. This must be set to a unique integer for each broker.
    broker.id=[0,1,2]
    
    ############################# Socket Server Settings #############################
    
    listeners=PLAINTEXT://CentOS[A,B,C]:9092
    
    ############################# Log Basics #############################
    
    log.dirs=/usr/kafka-logs
    
    ############################# Log Retention Policy #############################
    
    log.retention.hours=168
    
    ############################# Zookeeper #############################
    
    zookeeper.connect=CentOSA:2181,CentOSB:2181,CentOSC:2181
    
    [root@CentOSX kafka_2.11-2.2.0]#./bin/kafka-server-start.sh -daemon config/server.properties

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODIzMTQ0OA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200104182427750.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODIzMTQ0OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20201025134651891.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODIzMTQ0OA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200104182455113.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODIzMTQ0OA==,size_16,color_FFFFFF,t_70
[img-sScYm287-1603547057588_assets_20200104182508649.png]: https://img-blog.csdnimg.cn/20201025135444987.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-h17kZwGD-1603547057590_assets_20200104182529801.png]: https://img-blog.csdnimg.cn/20201025135510224.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-aKbGYdJW-1603547057593_assets_20200104182545392.png]: https://img-blog.csdnimg.cn/20201025135534718.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-Df1WeqBg-1603547057599_assets_20200104182620255.png]: https://img-blog.csdnimg.cn/2020102513560155.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-CLJAGOSH-1603547057602_assets_20200104182633932.png]: https://img-blog.csdnimg.cn/20201025135941292.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-m9UTim2q-1603547057607_assets_20200104182654616.png]: https://img-blog.csdnimg.cn/20201025135627631.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-VBp7r8QY-1603547057608_assets_20200104182707774.png]: https://img-blog.csdnimg.cn/20201025140026409.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-vo93m1w5-1603547057610_assets_20200104182725874.png]: https://img-blog.csdnimg.cn/20201025135700646.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-UyoNwYZW-1603547057613_assets_20200104182753258.png]: https://img-blog.csdnimg.cn/2020102513572563.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[img-yywvsD8C-1603547057616_assets_image-20201019120149718.png]: https://img-blog.csdnimg.cn/20201025135804954.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center