【头歌实训】kafka-入门篇

文章目录

第1关：kafka - 初体验
- 任务描述
- 相关知识
- - Kafka 简述
  - Kafka 应用场景
  - Kafka 架构组件
  - kafka 常用命令
- 编程要求
- 测试说明
- 答案代码
第2关：生产者（Producer ） - 简单模式
- 任务描述
- 相关知识
- - Producer 简单模式
  - Producer 的开发步骤
  - Kafka 常用配置参数
- 编程要求
- 测试说明
- 答案代码
第3关：消费者（ Consumer）- 自动提交偏移量
- 任务描述
- 相关知识
- - Kafka 消费者开发步骤
  - 自动提交偏移量的优劣
- 编程要求
- 测试说明
- 答案代码
第4关：消费者（ Consumer ）- 手动提交偏移量
- 任务描述
- 相关知识
- - Kafka 两种手动提交方式
- 编程要求
- 测试说明
- 答案代码

第1关：kafka - 初体验

任务描述

本关任务：使用 Kafka 命令创建一个副本数量为1、分区数量为3的 Topic 。

编程要求

根据提示，在右侧编辑器补充代码完成以下任务。

创建一个副本数量为1、分区数量为3、名为 demo 的 Topic
查看所有 Topic
查看名为 demo 的 Topic 的详情信息

注意：Broker 的 IP 为127.0.0.1，Zookeeper 的 IP 为127.0.0.1

扩展任务：

使用一个命令行开启 Kafka Producer Shell 窗口并对名为 demo 的 Topic 进行数据生产
使用另一个命令行开启 Kafka Customer Shell 窗口并对名为 demo 的 Topic进行消费

说明：扩展任务没有进行评测，此任务目的是为了让用户体验下 Kafka 的数据生产与数据消费的两个环节，更好理解 Kafka

测试说明

平台会对你的命令进行检验并运行，你只需要按照任务需求，补充右侧编辑器的代码，然后点击评测就ok了。 - -

特别注意：为了确保运行拥有一个正常的运行环境，请在评测之前，重置下运行环境

答案代码

命令行代码

kafka-server-start.sh  /opt/kafka_2.11-1.1.0/config/server.properties

shell 文件

#!/bin/bash
 
#1.创建一个副本数量为1、分区数量为3、名为 demo 的 Topic
kafka-topics.sh --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 3 --topic demo
 
#2.查看所有Topic
kafka-topics.sh --list --zookeeper  127.0.0.1:2181
 
#3.查看名为demo的Topic的详情信息
kafka-topics.sh --topic demo --describe --zookeeper 127.0.0.1:2181

第2关：生产者（Producer ） - 简单模式

任务描述

本关任务：编写一个 Kafka 的 Producer 进行数据生产。

名称	说明	默认值	有效值	重要性
bootstrap.servers	kafka集群的broker-list，如： hadoop01:9092,hadoop02:9092	无		必选
key.serializer	key的序列化器		ByteArraySerializer StringSerializer	必选
value.serializer	value的序列化器		ByteArraySerializer StringSerializer	必选
acks	确保生产者可靠性设置，有三个选项： acks=0：不等待成功返回 acks=1：等Leader写成功返回 acks=all：等Leader和所有ISR中的Follower写成功返回，all也可以用-1代替	-1	0,1,-1,all	建议必选
buffer.memory	Producer总体内存大小	33554432	不要超过物理内存，根据实际情况调整	建议必选
batch.size	每个partition的未发送消息大小	16384	根据实际情况调整	建议必选

编程要求

根据提示，在右侧编辑器补充代码，使用 Kafka Producer API 对名为 demo 的 Topic 进行数据生产。

测试说明

平台会对你的命令进行检验并运行，你只需要按照任务需求，补充右侧编辑器的代码，然后点击评测就ok了。 - -

答案代码

用 conf/server.properties ，如果用 config/server.properties 的话需要把 log.dirs 和 num.partitions 这两个配置改了

cd $KAFKA_HOME/
vim config/server.properties

# 使用 :/log.dirs 找到位置，或者直接 :$ 在最后一行加 
log.dirs=/export/servers/logs/kafka/
num.partitions=2

命令行执行代码

# kafka 依赖 zookeeper，所以需要先启动 zookeeper 服务
cd $ZOOKEEPER_HOME/
bin/zkServer.sh start conf/zoo.cfg
# 启动 Kafka 服务
cd $KAFKA_HOME/
bin/kafka-server-start.sh -daemon conf/server.properties

Java 代码

package net.educoder;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
 * kafka producer 简单模式
 */
public class App {
    public static void main(String[] args) {
        /**
         * 1.创建配置文件对象，一般采用 Properties
         */
        /**----------------begin-----------------------*/
        Properties props = new Properties();
        /**-----------------end-------------------------*/
        /**
         * 2.设置kafka的一些参数
         *          bootstrap.servers --> kafka的连接地址 127.0.0.1:9092
         *          key、value的序列化类 -->org.apache.kafka.common.serialization.StringSerializer
         *          acks：1，-1,0
         */
        /**-----------------begin-----------------------*/
        props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("acks", "1");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        /**-----------------end-------------------------*/
        /**
         * 3.构建kafkaProducer对象
         */
        /**-----------------begin-----------------------*/
        Producer<String, String> producer = new KafkaProducer<>(props);
        /**-----------------end-------------------------*/
        for (int i = 0; i < 100; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("demo", i + "", i + "");
            /**
             * 4.发送消息
             */
            /**-----------------begin-----------------------*/
            producer.send(record);
            /**-----------------end-------------------------*/
        }
        producer.close();
    }
}

第3关：消费者（ Consumer）- 自动提交偏移量

任务描述

本关任务：编写一个 Kafka 消费者并设置自动提交偏移量进行数据消费。

编程要求

使用 Kafka Consumer API 对名为 demo 的 Topic 进行消费，并设置自动提交偏移量。

测试说明

平台会对你的命令进行检验并运行，你只需要按照任务需求，补充右侧编辑器的代码，然后点击评测就 ok 了。 - -

答案代码

package net.educoder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class App {
    public static void main(String[] args) {
        Properties props = new Properties();
        /**--------------begin----------------*/
        //1.设置kafka集群的地址
        props.put("bootstrap.servers", "127.0.0.1:9092");
        //2.设置消费者组，组名字自定义，组名字相同的消费者在一个组
        props.put("group.id", "g1");
        //3.开启offset自动提交
        props.put("enable.auto.commit", "true");
        //4.自动提交时间间隔
        props.put("auto.commit.interval.ms", "1000");
        //5.序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        /**---------------end---------------*/
        /**--------------begin----------------*/
        //6.创建kafka消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //7.订阅kafka的topic
        consumer.subscribe(Arrays.asList("demo"));
        /**---------------end---------------*/
        int i = 1;
        while (true) {
            /**----------------------begin--------------------------------*/
            //8.poll消息数据,返回的变量为crs
            ConsumerRecords<String, String> crs = consumer.poll(100);
            for (ConsumerRecord<String, String> cr : crs) {
                System.out.println("consume data:" + i);
                i++;
            }
            /**----------------------end--------------------------------*/
            if (i > 10) {
                return;
            }
        }
    }
}

第4关：消费者（ Consumer ）- 手动提交偏移量

任务描述

本关任务：编写一个 Kafka 消费者并使用手动提交偏移量进行数据消费。

编程要求

根据提示，在右侧编辑器补充代码，使用 Kafka Producer API 对名为 demo 的 Topic 进行数据生产

测试说明

平台会对你的命令进行检验并运行，你只需要按照任务需求，补充右侧编辑器的代码，然后点击评测就ok了。 - -

答案代码

命令行代码

# kafka 依赖 zookeeper，所以需要先启动 zookeeper 服务
cd $ZOOKEEPER_HOME/
bin/zkServer.sh start conf/zoo.cfg
# 启动 Kafka 服务
cd $KAFKA_HOME/
bin/kafka-server-start.sh -daemon conf/server.properties

Java 代码

package net.educoder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
public class App {
    public static void main(String[] args){
        Properties props = new Properties();
        /**-----------------begin------------------------*/
        //1.设置kafka集群的地址
        props.put("bootstrap.servers", "127.0.0.1:9092");
        //2.设置消费者组，组名字自定义，组名字相同的消费者在一个组
        props.put("group.id", "g1");
        //3.关闭offset自动提交
        props.put("enable.auto.commit", "false");
        //4.序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        /**-----------------end------------------------*/
        /**-----------------begin------------------------*/
        //5.实例化一个消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //6.消费者订阅主题，订阅名为demo的主题
        consumer.subscribe(Arrays.asList("demo"));
        /**-----------------end------------------------*/
        final int minBatchSize = 10;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                buffer.add(record);
            }
            if (buffer.size() >= minBatchSize) {
                for (ConsumerRecord bf : buffer) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", bf.offset(), bf.key(), bf.value());
                }
                /**-----------------begin------------------------*/
                //7.手动提交偏移量
                consumer.commitSync();
                /**-----------------end------------------------*/
                buffer.clear();
                return;
            }
        }
    }
}