Kafka集群分布式部署与测试 —— 实操版

àì夳堔傛蜴生んèń 2023-01-18 13:24 8阅读 0赞

### Kafka集群分布式部署与测试 ###

*   *  一、Kafka 安装与启动
     *   *  1.1 创建文件
         *  1.2 解压文件
         *  1.3 配置环境变量（三个节点）
         *  1.4 分发文件，且修改配置文件server.properties(两个从节点)
         *  1.5 启动 kafka（三个节点）
     *  二、Kafka生产者和消费者实例
     *   *  2.1 创建 kafka的 主题
         *  2.2 master 创建生产者生产消息
         *  2.3 slave1 创建消费者消费信息
     *  三、基于java API 方式使用 Kafka
     *   *  3.1 添加依赖
         *  3.2 编写生产者客户端
         *  3.3 编写消费者客户端

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

**Kafka 的资源下载链接：**[https://pan.baidu.com/s/1WlWf\_\_Y8c0jXCOVAFixFRw][https_pan.baidu.com_s_1WlWf_Y8c0jXCOVAFixFRw]  
提取码：**v0lx**  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70][]

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

[Kafka 基础知识理论版][Kafka]

[Kafka Streams 小实践 ——开放词频统计应用][Kafka Streams _]

[Kafka集群分布式部署与测试 —— 实操版][Kafka_ _]

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

**目的：**

1.  学会配置Kafka ；
2.  学会使用 Kafka 生产者和消费者实例；
3.  基于java API 方式使用 Kafka。

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

## 一、Kafka 安装与启动 ##

### 1.1 创建文件 ###

`mkdir -p /usr/kafka`

### 1.2 解压文件 ###

`tar -zxvf kafka_2.11-2.4.0.tgz -C /usr/kafka/`

进入配置目录：  
`cd /usr/kafka/kafka_2.11-2.4.0/config`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 1]  
修改文件 **vi server.properties**：

# broker的全局唯一编号，不能重复，依次 0、1、2。
    broker.id=0
    # 用来监听链接的端口，producer或consumer将在此端口建立连接
    port=9092
    # 处理网络请求的线程数量
    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消息存放的路径
    log.dirs=/usr/kafka/kafka_2.11-2.4.0/tmp/kafka-logs
    # topic在当前broker上的分片个数
    num.partitions=2
    # 用来恢复和清理data下数据的线程数量
    num.recovery.threads.per.data.dir=1
    # segment文件保留的最长时间，超时将被删除
    log.retention.hours=168
    # 滚动生成新的segment文件的最大时间
    log.roll.hours=1
    # 日志文件中每个segment的大小，默认为1G
    log.segment.bytes=1073741824
    # 周期性检查文件大小的时间
    log.retention.check.interval.ms=300000
    # 日志清理是否打开
    log.cleaner.enable=true

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 2]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 5]

# broker需要使用zookeeper保存meta数据
    zookeeper.connect=master:2181,slave1:2181,slave2:2181
    
    # zookeeper链接超时时间
    zookeeper.connection.timeout.ms=6000
    # partion buffer中，消息的条数达到阈值，将触发flush到磁盘
    log.flush.interval.messages=10000
    # 消息buffer的时间，达到阈值，将触发flush到磁盘
    log.flush.interval.ms=3000
    # 删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除
    delete.topic.enable=true
    # host.name为本机IP,不改则会Producerconnection to localhost:9092 unsuccessful 错误!
    # host.name=slave1（slave2）
    host.name=master

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 6]  
修改完毕，保存退出。

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

### 1.3 配置环境变量（三个节点） ###

vi /etc/profile
    # kafka
    export KAFKA_HOME=/usr/kafka/kafka_2.11-2.4.0
    export PATH=$PATH:$KAFKA_HOME/bin
    source /etc/profile

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

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

### 1.4 分发文件，且修改配置文件server.properties(两个从节点) ###

![在这里插入图片描述][20210506164951853.png]  
![在这里插入图片描述][2021050616500356.png]![在这里插入图片描述][20210506165008668.png]  
同理，slave2也是这样。

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

### 1.5 启动 kafka（三个节点） ###

注意的是：要显启动 Zookeeper 集群服务，最好把hadoop也启动起来。

cd /usr/kafka/kafka_2.11-2.4.0/
    bin/kafka-server-start.sh config/server.properties

![在这里插入图片描述][20210506165127449.png]![在这里插入图片描述][20210506165133588.png]  
![在这里插入图片描述][20210506165935288.png]

再打开另一个shell master窗口，查看jps。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 8]

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

## 二、Kafka生产者和消费者实例 ##

### 2.1 创建 kafka的 主题 ###

在 Kafka 的 bin 目录下有 **kafka-topics.sh** 文件，通过该文件就可以操作与主题相关的功能，下面我们创建一个 **ittheme** 主题，分区数为3，副本数2，命令如下：

# 1.创建 ittheme 主题
    kafka-topics.sh --create --topic ittheme --partitions 3 --replication-factor 2 --zookeeper master:2181,slave1:2181,slave2:2181

### 2.2 master 创建生产者生产消息 ###

主题创建成功后，在master 创建生产者生产消息，通过bin目录下的 **kafka-console-producer.sh** 文件操作 —— 向主题发送消息数据，命令如下：

# 2.创建 ittheme 生产者
    kafka-console-producer.sh --broker-list master:9092,slave1:9092,slave2:9092 --topic ittheme

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

### 2.3 slave1 创建消费者消费信息 ###

接着在 slave1 创建消费者消费信息，通过 **kafka-console-consumer.sh** 文件读取ittheme主题内容。命令如下：

# 3.创建ittheme 消费者
    kafka-console-consumer.sh --from-beginning  --topic ittheme --bootstrap-server master:9092,slave1:9092,slave2:9092

![在这里插入图片描述][2021050617033257.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 9]

# --list查看主题，
    kafka-topics.sh --list --zookeeper master:2181,slave1:2181,slave2:2181
    
    # --delete删除主题
    kafka-topics.sh --delete --zookeeper master:2181,slave1:2181,slave2:2181 --topic theme

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

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

## 三、基于java API 方式使用 Kafka ##

用户不仅能够通过命令行形式操作 Kafka 服务，还能用多种语言操作 Kafka。用户在开发独立项目时，通过调用Kafka API来操作Kafka集群，其核心API主要有5种分别是：**Producer API、Consumer API 、Streams API、Connect API 、AdminClient API**。

**KafkaProducer API 常用API表**

<table> 
 <thead> 
  <tr> 
   <th>方法名称</th> 
   <th align="left">相关说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>abortTransaction()</td> 
   <td align="left">终止正在进行的事物；</td> 
  </tr> 
  <tr> 
   <td>close()</td> 
   <td align="left">关闭这个生产者；</td> 
  </tr> 
  <tr> 
   <td>flush()</td> 
   <td align="left">调用此方法使所有缓冲的记录立即发送；</td> 
  </tr> 
  <tr> 
   <td>partitionsFor(java.lang.String topic)</td> 
   <td align="left">获取给定主题的分区元数据；</td> 
  </tr> 
  <tr> 
   <td>send (ProducerRecord &lt;K,V&gt; record)</td> 
   <td align="left">异步发送记录到主题。</td> 
  </tr> 
 </tbody> 
</table>

**KafkaComsumer API 常用API表**

<table> 
 <thead> 
  <tr> 
   <th>方法名称</th> 
   <th align="left">相关说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>subscribe(java.util.Collection&lt;java.lang.String&gt; topics)</td> 
   <td align="left">订阅给定主题列表以获取动态分区;</td> 
  </tr> 
  <tr> 
   <td>close()</td> 
   <td align="left">关闭这个消费者;</td> 
  </tr> 
  <tr> 
   <td>wakeup()</td> 
   <td align="left">唤醒消费者；</td> 
  </tr> 
  <tr> 
   <td>metrics()</td> 
   <td align="left">获取消费者保留的指标；</td> 
  </tr> 
  <tr> 
   <td>listTopics()</td> 
   <td align="left">获取有关用户有权查看的所有主题的分区的元数据。</td> 
  </tr> 
 </tbody> 
</table>

### 3.1 添加依赖 ###

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

### 3.2 编写生产者客户端 ###

for循环 输入50个数字到消费者客户端

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    public class KafkaProducerTest {
        
        public static void main(String[] args) {
        
            Properties props = new Properties();
            // 1、指定Kafka集群的主机名和端口号
            props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
            // 2、指定等待所有副本节点的应答
            props.put("acks", "all");
            // 3、指定消息发送最大尝试次数
            props.put("retries", 0);
            // 4、指定一批消息处理大小
            props.put("batch.size", 16384);
            // 5、指定请求延时
            props.put("linger.ms", 1);
            // 6、指定缓存区内存大小
            props.put("buffer.memory", 33554432);
            // 7、设置key序列化
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            // 8、设置value序列化
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            // 9、生产数据
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
            for (int i = 0; i < 50; i++) {
        
                producer.send(new ProducerRecord<String, String>("theme", Integer.toString(i), "hello world-" + i));
            }
            producer.close();
        }
    }

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

### 3.3 编写消费者客户端 ###

接收生产者客户端信息，并读取内容显示。

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.clients.producer.Callback;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    public class KafkaConsumerTest {
        
        public static void main(String[] args) {
        
            // 1、准备配置文件
            Properties props = new Properties();
            // 2、指定Kafka集群主机名和端口号
            props.put("bootstrap.servers", "master:9092,slave1:9092,slave1:9092");
            // 3、指定消费者组ID，在同一时刻同一消费组中只有一个线程可以去消费一个分区数据，不同的消费组可以去消费同一个分区的数据。
            props.put("group.id", "theme");
            // 4、自动提交偏移量
            props.put("enable.auto.commit", "true");
            // 5、自动提交时间间隔，每秒提交一次
            props.put("auto.commit.interval.ms", "1000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
            // 6、订阅数据，这里的topic可以是多个
            kafkaConsumer.subscribe(Arrays.asList("theme"));
            // 7、获取数据
            while (true) {
        
                //每隔100ms就拉去一次
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
        
                    System.out.printf("topic = %s,offset = %d, key = %s, value = %s%n", record.topic(), record.offset(), record.key(), record.value());
                }
            }
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 11]  
实操完毕

[https_pan.baidu.com_s_1WlWf_Y8c0jXCOVAFixFRw]: https://pan.baidu.com/s/1WlWf__Y8c0jXCOVAFixFRw
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70]: /images/20221021/731b507d20724689baa7f5973aa67e5e.png
[Kafka]: https://blog.csdn.net/weixin_44775255/article/details/116465983
[Kafka Streams _]: https://blog.csdn.net/weixin_44775255/article/details/116458302
[Kafka_ _]: https://blog.csdn.net/weixin_44775255/article/details/116457242
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 1]: /images/20221021/6f3c08cd1e5e48e29a9b94a91cbe0689.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210506164631108.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 3]: /images/20221021/4c713b5bb93f47b582f779af5eba9df5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 4]: /images/20221021/7ec568a958034a9ba548d02a9cd1fa47.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 5]: /images/20221021/d8d94b8fb302434985ec53c234d2a951.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 6]: /images/20221021/ab750bfaed6d41909346bb91bedd4d02.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 7]: /images/20221021/beca5906aa8a4146aca776d31d69b341.png
[20210506164951853.png]: /images/20221021/b3b9f47e942045128c7813b5a244cb4f.png
[2021050616500356.png]: https://img-blog.csdnimg.cn/2021050616500356.png
[20210506165008668.png]: /images/20221021/ec4ed5b3ccea402cb83ef189b08d998b.png
[20210506165127449.png]: https://img-blog.csdnimg.cn/20210506165127449.png
[20210506165133588.png]: /images/20221021/007d4699f5d14561b03c36d75850643a.png
[20210506165935288.png]: /images/20221021/42ccc6984b034a04976e09dd176ea000.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 8]: /images/20221021/cb0cb038ddb043079048fc2b0ce42601.png
[20210506170320874.png]: /images/20221021/aea107f4658c4208842099254dda9722.png
[2021050617033257.png]: /images/20221021/5b89aa713d56412e9fa9c93c01a8a4da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 9]: /images/20221021/cffdaa050e604fa184534cd04e43f17d.png
[2021050617040497.png]: /images/20221021/1ff3875ac906467388bbfbc85bf17147.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 10]: /images/20221021/f0a92467ea934fa288a41eccfcf5abcb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 11]: /images/20221021/cb86e5b2f6d646acbd7d3834567e29fc.png