大数据-61 Kafka 高级特性消息消费02-主题与分区自定义反序列化拦截器位移提交位移管理重平衡

news2024/9/19 16:12:52

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（正在更新…）

章节内容

上节我们完成了如下内容：

消费组测试，消费者变动对消费的影响
消费者的心跳机制
消费者的相关配置参数

在这里插入图片描述

主题和分区

Topic：Kafka用于分类管理消息的逻辑单元，类似于MySQL的数据库
Partition：是Kafka下数据存储的基本单元，这个是物理上的概念，同一个Topic的数据，会被分散的存储到多个Partition中，这些Partition可以在同一台机器上，也可以在多台机器上。优势在于可以进行水平扩展，通常Partition的数量是BrokerServer数量的整数倍
ConsumerGroup，同样是逻辑上的概念，是Kafka实现单播和广播两种消息模型的手段。保证一个消费组获取到特定主题的全部消息。在消息组内部，若干个消费者消费主题分区的消息，消费组可以保证一个主题的每个分区只被消费组中的一个消费者消费。
Consumer 采用 PULL 模式从 Broker 中读取数据，采用PULL模式 Consumer可以自行控制消费的速度。

反序列化

Kafka的Broker中所有的消息都是字节数组，消费者获取到消息之后，需要先对消息进行反序列化处理，然后才能交由给用户程序消费。
消费者的反序列化器包括Key和Value。

自定义反序列化

如果要实现自定义的反序列化器，需要实现 Deserializer 接口：

public class UserDeserializer implements Deserializer<User> {


    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        Deserializer.super.configure(configs, isKey);
    }

    @Override
    public User deserialize(String topic, byte[] data) {
        ByteBuffer buffer = ByteBuffer.allocate(data.length);
        buffer.put(data);
        buffer.flip();
        int userId = buffer.getInt();
        int usernameLen = buffer.getInt();
        String username = new String(data, 8, usernameLen);
        int passwordLen = buffer.getInt();
        String password = new String(data, 8 + usernameLen, passwordLen);
        int age = buffer.getInt();
        User user = new User();
        user.setUserId(userId);
        user.setUsername(username);
        user.setPassword(password);
        user.setAge(age);
        return user;
    }

    @Override
    public User deserialize(String topic, Headers headers, byte[] data) {
        return Deserializer.super.deserialize(topic, headers, data);
    }

    @Override
    public void close() {
        Deserializer.super.close();
    }
}

消费者拦截器

消费者在拉取了分区消息之后，要首先经过反序列化器对Key和Value进行反序列化操作。
消费端定义消息拦截器，要实现 ConsumerInterceptor接口：

一个可插拔的接口，允许拦截、更改消费者接收到的消息，首要的用例在于将第三方组件引入消费者应用程序，用于定制监控、日志处理等
该接口的实现类通过configure方法获取消费者配置的属性，如果消费者配置中没有指定ClientID，还可以获取KafkaConsumer生成的ClientID，获取这个配置跟其他拦截器是共享的，需要保证不会在各个拦截器之间产生冲突。
ConsumerInterceptor方法抛出异常会被捕获，但不会向下传播，如果配置了错误的参数类型，消费者不会抛出异常而是记录下来。
ConsumerInterceptor回调发生在KafkaConsumer.poll()方法的同一个线程

public class ConsumerInterceptor01 implements ConsumerInterceptor<String, String> {

    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        System.out.println("=== 消费者拦截器 01 onConsume ===");
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        System.out.println("=== 消费者拦截器 01 onCommit ===");
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
        System.out.println("消费者设置的参数");
        configs.forEach((k, v) -> {
            System.out.println(k + ", " + v);
        });
    }
}