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

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

2024-10-14 6

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

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

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

目前已经更新到了：

Hadoop（已更完）

HDFS（已更完）

MapReduce（已更完）

Hive（已更完）

Flume（已更完）

Sqoop（已更完）

Zookeeper（已更完）

HBase（已更完）

Redis （已更完）

Kafka（正在更新…）

章节内容

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

消费组测试，消费者变动对消费的影响

消费者的心跳机制

消费者的相关配置参数

682ff08a122671ae7e960a1d0948a640_72d9547a846a4eacb683562672ba16c6.png 主题和分区

Topic：Kafka用于分类管理消息的逻辑单元，类似于MySQL的数据库

Partition：是Kafka下数据存储的基本单元，这个是物理上的概念，同一个Topic的数据，会被分散的存储到多个Partition中，这些Partition可以在同一台机器上，也可以在多台机器上。优势在于可以进行水平扩展，通常Partition的数量是BrokerServer数量的整数倍

ConsumerGroup，同样是逻辑上的概念，是Kafka实现单播和广播两种消息模型的手段。保证一个消费组获取到特定主题的全部消息。在消息组内部，若干个消费者消费主题分区的消息，消费组可以保证一个主题的每个分区只被消费组中的一个消费者消费。

Consumer 采用 PULL 模式从 Broker 中读取数据，采用PULL模式 Consumer可以自行控制消费的速度。

d938a1033a8c01828ed8c7f30fdda370_8883489da0f4494a92552b729d20e10f.png

反序列化

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);
        });
    }
}