Kafka（一）------Linux环境下Kafka集群安装

￡神魔★判官ぃ 2023-02-17 12:25 46阅读 0赞

## 一、下载 ##

Kafka下载地址：`http://kafka.apache.org/downloads.html`

https://download.csdn.net/download/qq\_29914837/12521366

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

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

## 2、安装kafka ##

前提：安装jdk，建议版本1.8

#解压至/usr/local/kafka/
    tar -zxvf kafka_2.12-2.5.0.tgz -C /usr/local/
    #切换目录
    cd /usr/local/
    # 重命名
    mv kafka_2.12-2.5.0 kafka

![在这里插入图片描述][2020061405575962.png]  
修改server.properties配置文件

vi config/server.properties

修改内容如下

#  broker就是一个kafka的部署实例，在一个kafka集群中，每一台kafka都要有一个broker.id
    #  并且，该id唯一，且必须为整数
    broker.id=1
    # 一个用逗号分隔的目录列表，用于存储日志文件
    log.dirs=/data/kafka-logs

server.properties 完整文件

############################# Server Basics #############################
    
    #  broker就是一个kafka的部署实例，在一个kafka集群中，每一台kafka都要有一个broker.id
    #  并且，该id唯一，且必须为整数
    broker.id=1
    
    ############################# Socket Server Settings #############################
    
    # The address the socket server listens on. It will get the value returned from 
    # java.net.InetAddress.getCanonicalHostName() if not configured.
    #   FORMAT:
    #     listeners = listener_name://host_name:port
    #   EXAMPLE:
    #     listeners = PLAINTEXT://your.host.name:9092
    #listeners=PLAINTEXT://:9092
    
    # Hostname and port the broker will advertise to producers and consumers. If not set, 
    # it uses the value for "listeners" if configured.  Otherwise, it will use the value
    # returned from java.net.InetAddress.getCanonicalHostName().
    #advertised.listeners=PLAINTEXT://your.host.name:9092
    
    # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
    #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
    
    # 默认处理网络请求的线程个数 3个
    num.network.threads=3
    
    # 执行磁盘IO操作的默认线程个数 8
    num.io.threads=8
    
    # socket服务使用的进行发送数据的缓冲区大小，默认100kb
    socket.send.buffer.bytes=102400
    
    # socket服务使用的进行接受数据的缓冲区大小，默认100kb
    socket.receive.buffer.bytes=102400
    
    # 套接字服务器将接受的请求的最大大小(针对OOM  (Out of memory)内存溢出)的保护)
    socket.request.max.bytes=104857600
    
    
    ############################# Log Basics #############################
    
    # A comma separated list of directories under which to store log files
    一个用逗号分隔的目录列表，用于存储日志文件
    log.dirs=/data/kafka-logs
    
    # 每一个topic所对应的log的partition分区数目，默认1个。更多的partition数目会提高消费
    # 并行度，但是也会导致在kafka集群中有更多的文件进行传输
    # （partition就是分布式存储，相当于是把一份数据分开几份来进行存储，即划分块、划分分区的意思）
    num.partitions=1
    
    # 每一个数据目录用于在启动kafka时恢复数据和在关闭时刷新数据的线程个数。如果kafka数据存储在磁盘阵列中
    # 建议此值可以调整更大。
    num.recovery.threads.per.data.dir=1
    
    ############################# Internal Topic Settings  #############################
    # The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
    # For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
    offsets.topic.replication.factor=1
    transaction.state.log.replication.factor=1
    transaction.state.log.min.isr=1
    
    ############################# Log Flush Policy 数据刷新策略#############################
    
    # Messages are immediately written to the filesystem but by default we only fsync() to sync
    # the OS cache lazily. The following configurations control the flush of data to disk.
    # There are a few important trade-offs here:
    #    1. Durability: Unflushed data may be lost if you are not using replication.
    #    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
    #    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
    # The settings below allow one to configure the flush policy to flush data after a period of time or
    # every N messages (or both). This can be done globally and overridden on a per-topic basis.
    
    # The number of messages to accept before forcing a flush of data to disk
    #log.flush.interval.messages=10000
    
    # The maximum amount of time a message can sit in a log before we force a flush
    #log.flush.interval.ms=1000
    
    ############################# Log Retention Policy #############################
    
    # The following configurations control the disposal of log segments. The policy can
    # be set to delete segments after a period of time, or after a given size has accumulated.
    # A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
    # from the end of the log.
    
    # The minimum age of a log file to be eligible for deletion due to age
    # 基于时间的策略，删除日志数据的时间，默认保存7天
    log.retention.hours=168
    
    # A size-based retention policy for logs. Segments are pruned from the log unless the remaining
    # segments drop below log.retention.bytes. Functions independently of log.retention.hours.
    #log.retention.bytes=1073741824
    
    # The maximum size of a log segment file. When this size is reached a new log segment will be created.
    # 数据分片策略
    log.segment.bytes=1073741824
    
    # The interval at which log segments are checked to see if they can be deleted according
    # to the retention policies
    # 每隔多长时间检测数据是否达到删除条件
    log.retention.check.interval.ms=300000
    
    ############################# Zookeeper #############################
    
    # Zookeeper connection string (see zookeeper docs for details).
    # This is a comma separated host:port pairs, each corresponding to a zk
    # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
    # You can also append an optional chroot string to the urls to specify the
    # root directory for all kafka znodes.
    zookeeper.connect=localhost:2181
    
    # Timeout in ms for connecting to zookeeper
    zookeeper.connection.timeout.ms=18000
    
    
    ############################# Group Coordinator Settings #############################
    
    # The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
    # The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
    # The default value for this is 3 seconds.
    # We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
    # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
    group.initial.rebalance.delay.ms=0

**PS:真正删除topic**

#删除 topic 功能使能
    delete.topic.enable=true
    #produce可以推送消息到一个不存在的topic（即：发消息到一个不存在的topic,系统会帮你按默认参数自动帮你建立这个topic）
    #所以只要这个topic有produce向它推送消息，那么这个topic你是没办法被删除的
    auto.create.topics.enable=false

## 三、安装启动zookeeper ##

PS:要先启动ZooKeeper再启动kafka，否则会报错

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

启动zookeeper分为内部和外部两种方式，即通过kafka自带zookeeper启动，或者另外安装zookeeper服务启动。

**3.1、通过kafka自带zookeeper启动**

cd /usr/local/kafka/kafka_2.12-2.5.0
    # 启动命令：守护进程进行启动
    bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

ps -ef|grep zookeeper

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

**3.2、单独安装zookeeper服务（推荐）**

[Zookeeper（二）------centos下zookeeper集群安装][Zookeeper_------centos_zookeeper]

## 四、启动kafka ##

cd /usr/local/kafka/kafka_2.12-2.5.0
    # 启动命令：守护进程进行启动
    bin/kafka-server-start.sh -daemon config/server.properties

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

查看服务启动情况

jps

新开的服务QuorumPeerMain是指zookeeper的进程。Kafka是指kafka的进程。  
![在这里插入图片描述][20200614064308553.png]

**停止kafka**

bin/kafka-server-stop.sh config/server.properties

## 五、Topic使用 ##

**5.1、创建topic ，名称为test**

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

选项说明：

–topic ： topic名称  
–replication ： 副本数  
–partitions : 分区数

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

**5.2、 查询topic列表**

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

![在这里插入图片描述][20200614071806133.png]  
**5.3、查看topic信息**

bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic test

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

**5.4、删除topic**

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic test

**5.5、启动生产者**

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

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

**5.6、启动消费者**

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

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

在最新的版本中，这种启动方式已经被删除了。

最新替换执行指令

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

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

ps: 建议开启两个终端，测试效果更明显，生产者发送消息，消费者实时消费消费

## 六、集群配置 ##

集群配置有两种方案

> 单机多broker集群配置；  
> 多机多broker集群配置

**6.1、单机多broker集群配置**

**复制server.properties文件**

cp config/server.properties config/server-1.properties
    cp config/server.properties config/server-2.properties

![在这里插入图片描述][20200615000035848.png]  
**修改配置文件信息**

#server-1.properities修改
    broker.id=1
    listeners=PLAINTEXT://192.168.1.204:9093
    log.dirs=/data/kafka-logs-1

#server-2.properities修改
    broker.id=2
    listeners=PLAINTEXT://192.168.1.204:9093
    log.dirs=/data/kafka-logs-2

broker.id=1
    
    listeners=PLAINTEXT://192.168.1.204:9093
    
    num.network.threads=3
    
    num.io.threads=8
    
    socket.send.buffer.bytes=102400
    
    socket.receive.buffer.bytes=102400
    
    socket.request.max.bytes=104857600
    
    log.dirs=/data/kafka-logs-1
    
    num.partitions=1
    
    num.recovery.threads.per.data.dir=1
    
    offsets.topic.replication.factor=1
    transaction.state.log.replication.factor=1
    transaction.state.log.min.isr=1
    
    log.retention.hours=168
    
    log.segment.bytes=1073741824
    
    log.retention.check.interval.ms=300000
    
    zookeeper.connect=localhost:2181
    
    zookeeper.connection.timeout.ms=18000
    
    group.initial.rebalance.delay.ms=0

**启动2个kafka服务**

bin/kafka-server-start.sh -daemon config/server-1.properties 
    bin/kafka-server-start.sh -daemon config/server-2.properties

**6.2、多机多broker集群配置**

<table> 
 <thead> 
  <tr> 
   <th>ip</th> 
   <th>Kafka</th> 
   <th>port</th> 
   <th>zookeeper</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>192.168.1.200</td> 
   <td>broker.id=1</td> 
   <td>9092</td> 
   <td>note1</td> 
  </tr> 
  <tr> 
   <td>192.168.1.201</td> 
   <td>broker.id=2</td> 
   <td>9092</td> 
   <td>note2</td> 
  </tr> 
  <tr> 
   <td>192.168.1.202</td> 
   <td>broker.id=3</td> 
   <td>9092</td> 
   <td>note3</td> 
  </tr> 
 </tbody> 
</table>

先在每台服务器安装kafka及zookeeper服务。

分别配置多个机器上的Kafka服务

设置不同的broke id
    更改对应ip

**针对192.168.1.200**

#节点id，用于区分子节点
    broker.id=1
    #指定主机地址，区分localhost，避免出错
    host.name=192.168.1.200
    #监听地址
    listeners = PLAINTEXT://192.168.1.200:9092
    #添加此项是为了api能够监听到此服务
    #advertised.listeners = PLAINTEXT://192.168.1.200:9092
     
    num.network.threads=3
    num.io.threads=8
    
    socket.send.buffer.bytes=102400
    socket.receive.buffer.bytes=102400
    socket.request.max.bytes=104857600
    
    #修改日志存储地址
    log.dirs=/data/kafka-logs
    
    num.partitions=1
    num.recovery.threads.per.data.dir=1
    
    offsets.topic.replication.factor=1
    transaction.state.log.replication.factor=1
    transaction.state.log.min.isr=1
    
    
    log.retention.hours=168
    log.segment.bytes=1073741824
    log.retention.check.interval.ms=300000
    
    #使用外部的zookeeper配置
    #zookeeper.connect=localhost:2181
    zookeeper.connect=192.168.1.200:2181,192.168.1.201:2181,192.168.1.202:2181
    zookeeper.connection.timeout.ms=18000
    
    group.initial.rebalance.delay.ms=0

**集群下测试topic**

#创建topic(test)
    bin/kafka-topics.sh --create --zookeeper 192.168.1.200:2181,192.168.1.201:2181,192.168.1.202:2181 --replication-factor 1 --partitions 1 --topic test
    #查看topic列表
    bin/kafka-topics.sh --list --zookeeper 192.168.1.200:2181,192.168.1.201:2181,192.168.1.202:2181
    #查看topic信息（test)
    bin/kafka-topics.sh --zookeeper 192.168.1.200:2181,192.168.1.201:2181,192.168.1.202:2181 --describe --topic test

200  
![在这里插入图片描述][20200615004336730.png]  
201  
![在这里插入图片描述][20200615004352452.png]

测试发现三台Kafka服务器，同步的，测试通过！

## 八、Kafka开机自启动 ##

cd /etc/rc.d/init.d/
    #创建kafka开机自启脚本文件
    vi kafka

#!/bin/sh
    #chkconfig: 2345 99 90
    #description: kafka
    #设置java安装路径
    export JAVA_HOME=/usr/java/jdk1.8.0_191
    #kafka auto start
    #切换到kafka的解压目录
    cd /usr/local/kafka/kafka_2.12-2.5.0/bin/
    #切换到kafka的解压目录
    echo "kafka start .............."
    ./kafka-server-start.sh -daemon ../config/server.properties
    echo "kafka end ................"

> \#chkconfig:2345 99 90 中100为优先级  
> number 越小，脚本的运行优先级就越高

这里我设置kafka文件中优先级为99，注意：zookeeper优先级一定要比kafka的优先级高，即数字更小。

#给文件添加权限
    chmod +x kafka
    #配置开机启动项
    chkconfig --add kafka
    #查看kafk配置是否成功
    chkconfig --list
    #关闭
    chkconfig --del kafka

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

参考资料

[kafka概述][kafka]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200614011237880.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200614013535572.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5OTE0ODM3,size_16,color_FFFFFF,t_70
[2020061405575962.png]: https://img-blog.csdnimg.cn/2020061405575962.png
[20200614062547319.png]: https://img-blog.csdnimg.cn/20200614062547319.png
[20200614062701617.png]: https://img-blog.csdnimg.cn/20200614062701617.png
[Zookeeper_------centos_zookeeper]: https://yilei.blog.csdn.net/article/details/106745554
[20200614062731192.png]: https://img-blog.csdnimg.cn/20200614062731192.png
[20200614064308553.png]: https://img-blog.csdnimg.cn/20200614064308553.png
[20200614070339648.png]: https://img-blog.csdnimg.cn/20200614070339648.png
[20200614071806133.png]: https://img-blog.csdnimg.cn/20200614071806133.png
[20200614071815755.png]: https://img-blog.csdnimg.cn/20200614071815755.png
[20200614130501528.png]: https://img-blog.csdnimg.cn/20200614130501528.png
[20200614130800866.png]: https://img-blog.csdnimg.cn/20200614130800866.png
[20200614130927984.png]: https://img-blog.csdnimg.cn/20200614130927984.png
[20200615000035848.png]: https://img-blog.csdnimg.cn/20200615000035848.png
[20200615004336730.png]: https://img-blog.csdnimg.cn/20200615004336730.png
[20200615004352452.png]: https://img-blog.csdnimg.cn/20200615004352452.png
[kafka]: https://www.w3cschool.cn/apache_kafka/apache_kafka_introduction.html