分布式实时消息队列Kafka（三）生产分区规则-阿里云开发者社区

分布式实时消息队列Kafka（三）

知识点01：课程回顾

请简述Kafka的集群架构及角色功能？

Kafka：分布式主从架构

主： Controller：管理集群中的Topic、分区、副本选举
从：Broker：对外接受读写请求，存储分区数据

Zookeeper

辅助选举Active的主节点：Crontroller
存储核心元数据

请简述Kafka中Topic管理的脚本及常用选项参数？

使用命令行中的脚本命令实现管理
脚本：kafka-topics.sh
常用选项

–topic
–create
–list
–describe
–delete
–alter：调整Topic的配置
–bootstrap-server / --broker-list

请简述如何使用Kafka Simple Java API 实现数据生产？描述具体的类及方法

step1：构建生产者连接对象：KafkaProducer

需要配置对象：管理配置，例如连接地址：Properties

step2：KafkaProducer：send：生产数据到Kafka中

需要构建一个生产的数据对象：ProducerRecord
ProducerRecord（Topic，Value）
ProducerRecord（Topic，Key，Value）
ProducerRecord（Topic，Partition，Key，Value）

请简述如何使用Kafka Simple Java API 实现数据消费？描述具体的类及方法

step1：构建消费者连接对象：KafkaConsumer

需要配置对象：管理配置，例如连接地址：Properties

step2：消费者需要订阅Topic

KafkaConsumer：subscribe（List）

step3：消费数据

KafkaConsumer：poll
ConsumerRecords：拉取到的所有数据
ConsumerRecord：消费到的每一条数据

topic
partition
offset
key
value

请简述Kafka生产数据时如何保证生产数据不丢失？

acks：返回的确认，当接收方收到数据以后，就会返回一个确认的消息
生产者向Kafka生产数据，根据配置要求Kafka返回ACK

ack=0：生产者不管Kafka有没有收到，直接发送下一条

优点：快
缺点：容易导致数据丢失，概率比较高

ack=1：生产者将数据发送给Kafka，Kafka等待这个分区leader副本写入成功，返回ack确认，生产者发送下一条

优点：性能和安全上做了平衡
缺点：依旧存在数据丢失的概率，但是概率比较小

ack=all/-1：生产者将数据发送给Kafka，Kafka等待这个分区所有副本全部写入，返回ack确认，生产者发送下一条

优点：数据安全
缺点：慢

如果Kafka没有返回ACK怎么办？

生产者会等待Kafka返回ACK，有一个超时时间，如果Kafka在规定时间内没有返回ACK，说明数据丢失了
生产者有重试机制，重新发送这条数据给Kafka
问题：如果ack在中途丢失，Kafkahi导致数据重复问题，怎么解决？

知识点02：课程目标

生产数据的分区规则？【重点】

常见的规则

MapReduce：Hash分区

优点：相同的Key会进入同一个分区

reduce0：part-r-00000

hadoop    6
spark   7

reduce1：part-r-00001

hbase   9
hive    10

缺点：数据倾斜的问题

如果所有Key的Hash取余的结果一样，导致数据分配不均衡的问题

Hbase：范围分区
轮询分区

优点：数据分配更加均衡
缺点：相同Key的数据进入不同的分区中

随机分区
槽位分区

有几种规则？
每种规则的优缺点是什么？

Kafka中消费数据如何保证不丢失不重复【重点】

Kafka中的消费者是如何消费数据？

消费者根据每个分区的Offset来进行消费
问题：数据丢失和数据重复的问题

如何解决数据安全的问题？

如果保证每次消费的offset是正确的offset？

知识点03：生产分区规则

目标：掌握Kafka生产者生产数据的分区规则
路径

问题：为什么生产数据的方式不同，分区的规则就不一样？

- ProducerRecord（Topic，Value）
- ProducerRecord（Topic，Key，Value）
- ProducerRecord（Topic，Partition，Key，Value）

step1：先判断是否指定了分区

- ProducerRecord（Topic，Partition，Key，Value）

step2：再判断是否给定了Key

指定了Key

- ProducerRecord（Topic，Key，Value）

没有指定Key

- ProducerRecord（Topic，Value）

实施

如果指定了分区的规则：写入所指定的分区中
如果没指定分区

默认调用的是DefaultPartitioner分区器中partition这个方法

如果指定了Key：按照Key的Hash取余分区的个数，来写入对应的分区
如果没有指定Key

2.x之前：轮询分区

优点：数据分配相对均衡

Topic   part    key   value
topic   0     1   itcast1
topic   1     2   itcast2
topic   2     3   itcast3
topic   0     4   itcast4
topic   1     5   itcast5
topic   2     6   itcast6
topic   0     7   itcast7
topic   1     8   itcast8
topic   2     9   itcast9

缺点：性能非常差

Kafka生产者写入数据：先将数据放入一个缓存中，与分区构建一个连接，发送一个批次的数据
第一条数据：先构建0分区的连接，第二条不是0分区的，所以直接构建一个批次，发送第一条
第二条数据：先构建1分区的连接，第三条不是1分区的，所以直接构建一个批次，发送第二条
……
每条数据需要构建一个批次，9条数据，9个批次，每个批次一条数据
批次多，每个批次数据量少，性能比较差
希望：批次少，每个批次数据量多，性能比较好

2.x之后：黏性分区

设计：让数据尽量的更加均衡，实现少批次多数据
规则

第一次：将所有数据随机选择一个分区，全部写入这个分区中，将这次的分区编号放入缓存中

bigdata01 1 37  null  itcast0
bigdata01 1 38  null  itcast1
bigdata01 1 39  null  itcast2
bigdata01 1 40  null  itcast3
bigdata01 1 41  null  itcast4
bigdata01 1 42  null  itcast5
bigdata01 1 43  null  itcast6
bigdata01 1 44  null  itcast7
bigdata01 1 45  null  itcast8
bigdata01 1 46  null  itcast9

第二次开始根据缓存中是否有上一次的编号

有：直接使用上一次的编号
如果没有：重新随机选择一个

小结

Kafka中生产数据的分区规则是什么？
是否指定了分区

就写入对应的分区

如果没有执行分区

是否指定了Key：Key的Hash取余分区
没有指定Key：黏性分区

尽量保证数据均衡前提下，实现少批次多数据

知识点04：自定义开发生产分区器

目标：掌握Kafka自定义开发生产分区器，以随机分区为例
路径

step1：开发一个类实现Partitioner接口
step2：实现partition方法
step3：生产者加载分区器

实施

开发一个随机分区器

package bigdata.itcast.cn.kafka.partition;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;
import java.util.Random;
/**
 * @ClassName UserPartition
 * @Description TODO 自定义分区器，实现随机分区
 * @Date 2021/3/31 9:21
 * @Create By     Frank
 */
public class UserPartition implements Partitioner {
    /**
     * 返回这条数据对应的分区编号
     * @param topic：Topic的名
     * @param key：key的值
     * @param keyBytes：key的字节
     * @param value：value的值
     * @param valueBytes：value的字节
     * @param cluster：集群对象
     * @return
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        //获取Topic的分区数
        Integer count = cluster.partitionCountForTopic(topic);
        //构建一个随机对象
        Random random = new Random();
        //随机得到一个分区编号
        int part = random.nextInt(count);
        return part;
    }
    @Override
    public void close() {
        //释放资源
    }
    @Override
    public void configure(Map<String, ?> configs) {
        //获取配置
    }
}

加载分区器

//指定分区器的类
props.put("partitioner.class","bigdata.itcast.cn.kafka.partition.UserPartition");

小结

如何构建一个自定义分区器
step1：构建一个类实现Partitioner接口
step2：实现partition方法：定义分区逻辑
step3：加载指定分区器即可

知识点05：消费者消费过程及问题

目标：掌握Kafka消费者消费过程及消费问题
路径

step1：消费者是如何消费Topic中的数据的？
step2：如果消费者故障重启，消费者怎么知道自己上次消费的位置的？

实施

Kafka中消费者消费数据的规则

消费者消费Kafka中的Topic根据Offset进行消费，每次从上一次的位置继续消费
第一次消费规则：由属性决定

auto.offset.reset=latest | earliest
latest：默认的值，从Topic每个分区的最新的位置开始消费
earliest：从最早的位置开始消费，每个分区的offset为0开始消费

第二次消费开始：根据上一次消费的Offset位置+1继续进行消费
问题1：消费者如何知道上一次消费的位置是什么？

每个消费者都将自己上一次消费的offset记录自己的内存中

问题2：如果因为网络资源原因，消费者故障了，重启消费者，原来内存中offset就没有了，消费者怎么知道上一次消费的位置？

Kafka Offset偏移量管理

Kafka将每个消费者消费的位置主动记录在一个Topic中：__consumer_offsets
如果下次消费者没有给定请求offset，kafka就根据自己记录的offset来提供消费的位置

提交的规则：根据时间自动提交

props.setProperty("enable.auto.commit", "true");//是否自动提交offset
props.setProperty("auto.commit.interval.ms", "1000");//提交的间隔时间

小结

消费者如何消费kafka中的Topic的数据

第一次消费：根据属性

auto.offset.reset：lastest/earliest

第二次消费开始：消费者在内存中记录上一次消费的offset + 1 = 这一次要消费的位置

问题：如果消费者故障，重启，不知道上一次的位置怎么办？

Kafka中记录了每个消费者这一次要消费的位置
由消费者定期的自动提交

知识点06：自动提交问题

目标：了解Kafka自动提交Offset存在的问题
路径

step1：自动提交是否会出现数据丢失问题
step2：自动提交是否会出现数据重复问题

实施

自动提交的规则

根据时间周期来提交下一次要消费的offset

props.setProperty("enable.auto.commit", "true");//是否自动提交offset
props.setProperty("auto.commit.interval.ms", "1000");//提交的间隔时间

数据丢失的情况

如果刚消费，还没处理，就达到提交周期，记录了当前的offset
最后处理失败，需要重启，重新消费处理
Kafka中已经记录消费过了，从上次消费的后面进行消费

数据重复的情况

如果消费并处理成功，但是没有提交offset，程序故障
重启以后，kafka中记录的还是之前的offset，重新又消费一遍
数据重复问题

小结

消费是否成功，是根据处理的结果来决定的
提交offset是根据时间来决定了
需要：根据处理的结果来决定是否提交offset

如果消费并处理成功：提交offset
如果消费处理失败：不提交offset

知识点07：实现手动提交Topic的Offset

目标：了解Kafka如何实现手动提交Topic的Offset实现
路径

step1：关闭自动提交
step2：消费完整后手动提交

实施

关闭自动提交

props.setProperty("enable.auto.commit", "false");//是否自动提交offset
//        props.setProperty("auto.commit.interval.ms", "1000");//提交的间隔时间

手动提交Offset

//处理完成以后，手动提交offset
            consumer.commitSync();

小结

关闭自动提交
根据处理的结果来实现手动提交

如果成功以后，再提交

知识点08：手动提交Topic Offset的问题

目标：了解Kafka实现手动提交Topic的Offset的问题
路径

step1：Offset的设计层次

Offset是什么级别的概念？

分区级别的概念

step2：手动提交Topic Offset出现数据重复问题
step3：解决方案是什么？

实施

Offset的设计

Offset是分区级别，每个分区单独管理一套offset

手动提交Topic Offset的过程中会出现数据重复

举个栗子
一个消费者，消费一个Topic，Topic有三个分区

第一次消费

part0

0 hadoop
1 hive

part1

0 hive
1 spark
2 hue

part2

0 spark
1 hadoop

问题：part0和part1都处理成功，当处理part2时候，程序故障，处理使用

offset有没有提交？没有提交
重启消费者：Kafka中没有消费记录，但是消费者刚刚分区0和分区1已经消费成功了
所有分区都重新消费

原因

Offset是分区级别的
提交offset是按照整个Topic级别来提交的

解决

提交offset的时候，按照分区来提交
消费成功一个分区，就提交一个分区的offset

小结

导致问题：数据重复
导致原因：offset是分区级别，提交时topic级别，只要有一个分区失败，整个提交失败，实际上部分分区已经处理成功了

知识点09：手动提交分区Offset的实现

目标：掌握Kafka实现手动提交Partition的Offset
路径

step1：消费每个分区的数据
step2：处理输出每个分区的数据
step3：手动提交每个分区的Offset

实施

获取所有数据中的分区

Set<TopicPartition> partitions = records.partitions();

从所有数据中取出每个分区的数据，输出，手动提交分区的Offset

//取出每个Partition的数据
            for (TopicPartition partition : partitions) {
                //将这个partition的数据从records中取出
                List<ConsumerRecord<String, String>> partRecords = records.records(partition);
                //遍历这个分区的每一条数据
                //取出每一条数据
                long offset = 0;
                for (ConsumerRecord<String, String> record : partRecords) {
                    //获取topic
                    String topic = record.topic();
                    //获取分区
                    int part= record.partition();
                    //获取offset
                    offset = record.offset();
                    //获取Key
                    String key = record.key();
                    //获取Value
                    String value = record.value();
                    System.out.println(topic+"\t"+part+"\t"+offset+"\t"+key+"\t"+value);
                }
                //分区数据处理结束，提交分区的offset
                Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(partition,new OffsetAndMetadata(offset+1));
                consumer.commitSync(offsets);
            }

观察结果

每个分区处理成功，就提交offset

小结

怎么实现基于分区提交offset
step1：获取所有数据
step2：获取所有分区
step3：处理每个分区的数据
step4：处理成功，提交分区处理的offset + 1
注意：工作中：一般不将Offset由Kafka存储，一般自己存储

如果处理成功：将offset存储在MySQL或者Zookeeper中
如果重启程序：从MySQL或者Zookeeper读取上一次的offset来实现

知识点10：指定消费Topic分区的数据

目标：掌握Kafka如何实现消费指定分区的数据
路径

step1：构建Topic分区对象
step2：指定消费Topic的分区
step3：输出消费结果

实施

构建Topic分区对象

//构建分区对象
TopicPartition part0 = new TopicPartition("bigdata01", 0);
TopicPartition part1 = new TopicPartition("bigdata01", 1);

指定消费Topic分区

//指定消费某些分区的数据
consumer.assign(Arrays.asList(part0,part1));

观察结果

小结

构建Topic的分区对象：TopicPartition
消费者指定消费分区：consumer.assign(Collection)

附录一：Maven依赖

<repositories>
    <repository>
        <id>aliyun</id>
        <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
    </repository>
</repositories>
<dependencies>
    <!-- Kafka的依赖 -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.4.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>2.4.1</version>
    </dependency>
</dependencies>

7943)]

小结

构建Topic的分区对象：TopicPartition
消费者指定消费分区：consumer.assign(Collection)

附录一：Maven依赖

<repositories>
    <repository>
        <id>aliyun</id>
        <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
    </repository>
</repositories>
<dependencies>
    <!-- Kafka的依赖 -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.4.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>2.4.1</version>
    </dependency>
</dependencies>

分布式实时消息队列Kafka（三）生产分区规则

分布式实时消息队列Kafka（三）

知识点01：课程回顾

知识点02：课程目标

知识点03：生产分区规则

知识点04：自定义开发生产分区器

知识点05：消费者消费过程及问题

知识点06：自动提交问题

知识点07：实现手动提交Topic的Offset

知识点08：手动提交Topic Offset的问题

知识点09：手动提交分区Offset的实现

知识点10：指定消费Topic分区的数据

附录一：Maven依赖

附录一：Maven依赖

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

分布式实时消息队列Kafka（三）生产分区规则

分布式实时消息队列Kafka（三）

知识点01：课程回顾

知识点02：课程目标

知识点03：生产分区规则

知识点04：自定义开发生产分区器

知识点05：消费者消费过程及问题

知识点06：自动提交问题

知识点07：实现手动提交Topic的Offset

知识点08：手动提交Topic Offset的问题

知识点09：手动提交分区Offset的实现

知识点10：指定消费Topic分区的数据

附录一：Maven依赖

附录一：Maven依赖

热门文章

最新文章

相关课程

相关电子书

相关实验场景