【头歌实训】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的topicconsumer.subscribe(Arrays.asList("demo"));/**---------------end---------------*/int i = 1;while (true) {/**----------------------begin--------------------------------*///8.poll消息数据,返回的变量为crsConsumerRecords<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;}}}
}