当前位置： 首页 > article >正文

Kafka 消费者专题

article 2025/1/7 16:16:33

目录

  • 消费者
    • 消费者组
    • 消费方式
    • 消费规则
    • 独立消费主题
      • 代码示例（极简）
      • 代码示例（独立消费分区）
    • offset
      • 自动提交
      • 代码示例（自动提交）
      • 手动提交
      • 代码示例（同步）
      • 代码示例（异步）
    • 其他说明

消费者

消费者组

  1. 由多个消费者组成
  2. 消费者组之间互不影响。
  3. 其他消费规则如下

消费方式

  1. push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。

  2. consumer采用pull（拉）模式从broker中读取数据。pull模式则可以根据consumer的消费能力以适当的速率消费消息。
    pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。

消费规则

在这里插入图片描述

  1. 一个消费者（单独消费者或消费者组中的一个）可以消费一个分区中的数据也可以消费两个或以上的分区数据
  2. 消费者组中的消费者必须访问不同的数据分区，不能访问同一个
  3. 同一个分区中的数据允许被不同的消费者访问（消费者不属于同一个组）

独立消费主题

代码示例（极简）

以下代码创建模拟一个消费者组（testCg）中的消费者，订阅来自topicA的消息
CustomTopicConsumer.java

package com.wunaiieq.consumer;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;


import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;


//创建一个独立消费者，消费topicA主题下的数据
public class CustomTopicConsumer {
    public static void main(String[] args) {
        //1.创建消费者属性文件对象
        Properties prop = new Properties();
        //2.为属性对象设置相关参数
        //设置kafka服务器
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.16.100:9092");
        //设置key和value的序列化类
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        //设置消费者的消费者组的名称
        prop.put(ConsumerConfig.GROUP_ID_CONFIG, "testCg");
        //3.创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer =
                new KafkaConsumer<String, String>(prop);
        //4.注册要消费的主题
        ArrayList<String> topics = new ArrayList<>();
        topics.add("topicA");
        //订阅主题
        kafkaConsumer.subscribe(topics);
        //5.拉取数据并打印输出
        while (true) {
            //6.设置1s消费一批数据
            ConsumerRecords<String, String> consumerRecords =
                    kafkaConsumer.poll(Duration.ofSeconds(1));
            //7.打印输出消费到的数据
            for (ConsumerRecord consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

代码示例（独立消费分区）

这个消费者需要消费topicA分区下的0号分区数据

package com.wunaiieq.consumer;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;


import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;


//创建一个独立消费者，消费topicA主题0号分区中的数据
public class ConsumTopicPartitionConsumer {
    public static void main(String[] args) {
        //1.创建属性对象
        Properties prop = new Properties();
        //2.设置相关参数
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.16.100:9092");
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        prop.put(ConsumerConfig.GROUP_ID_CONFIG,"testCg2");
        //3.创建消费者对象
        KafkaConsumer<String,String> kafkaConsumer =
                new KafkaConsumer<String, String>(prop);
        //4.为消费者注册主题和分区号
        List<TopicPartition> topicPartitions =
                new ArrayList<>();
        topicPartitions.add(new TopicPartition("topicA",0));
        kafkaConsumer.assign(topicPartitions);
        //5.消费数据
        while(true){
            ConsumerRecords<String, String> consumerRecords =
                    kafkaConsumer.poll(Duration.ofSeconds(1));
            for(ConsumerRecord consumerRecord:consumerRecords){
                System.out.println(consumerRecord);
            }
        }
    }
}

offset

表示消费者在特定主题分区中的消费进度。
一般而言，这个offset不会主动去用，除非宕机重启等等
可以手动查看offset值和状况

修改系统配置
[root@node4 ~]# cd /opt/kafka/config/ [root@node4 config]# vim
consumer.properties exclude.internal.topics=false
查询offset
kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server node2:9092 --consumer.config config/consumer.properties --formatter “kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter” --from-beginning

描述和作用：

  • Offset是Kafka中标识消息在分区内位置的一个唯一标识符。每个消息都有一个对应的Offset值，用于表示消息在分区中的相对位置。Offset是从0开始递增的，每当有新的消息写入分区时，Offset就会加1。Offset是不可变的，即使消息被删除或过期，Offset也不会改变或重用。
  • 定位消息：通过指定Offset，消费者可以准确地找到分区中的某条消息，或者从某个位置开始消费消息。
  • 记录消费进度：消费者在消费完一条消息后，需要提交Offset来告诉Kafka Broker自己消费到哪里了。这样，如果消费者发生故障或重启，它可以根据保存的Offset来恢复消费状态。
  • __consumer_offsets 主题里面采用 key 和 value 的方式存储数据。key 是 group.id+topic名称+分区号，value 就是当前 offset 的值。每隔一段时间，kafka 内部会对这个 topic 进行compact，也就是每个 group.id+topic名称+分区号就保留最新数据。

自动提交

自动提交主要是根据时间设置，每隔一段时间提交

代码示例（自动提交）

设置offset自动提交，每xxx秒提交一次（默认是5秒）

package com.wunaiieq.offset;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;


import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;


public class ConsumerAutoOffset {
    public static void main(String[] args) {
        //1.创建属性对象
        Properties prop = new Properties();
        //2.设置属性参数
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.16.100:9092");
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        //配置消费者组
        prop.put(ConsumerConfig.GROUP_ID_CONFIG,"cgauto");
        //是否自动提交offset： true表示自动提交，false表示非自动提交
        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
        //提交offset的时间周期1000ms,默认是5000ms
        prop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);
        //3.创建消费者对象
        KafkaConsumer<String,String> kafkaConsumer =
                new KafkaConsumer<String, String>(prop);
        //4.设置消费主题
        kafkaConsumer.subscribe(Arrays.asList("topicA"));
        //5.消费消息
        while(true){
            //6.读取消息
            ConsumerRecords<String, String> consumerRecords =
                    kafkaConsumer.poll(Duration.ofSeconds(1));
            //7.循环输出消息
            for(ConsumerRecord cr:consumerRecords){
                System.out.println(cr.value());
            }
        }
    }
}

手动提交

手动提交offset的方法有两种方式：

  1. commitSync同步提交:必须等待offset提交完毕，再去消费下一批数据。

  2. commitAsync异步提交:发送完提交offset请求后，就开始消费下一批数据了。
    两者的区别：

相同点是，都会将本次消费的一批数据的最高的偏移量提交；
不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而异步提交则没有失败重试机制，故有可能提交失败。

代码示例（同步）

等待offset提交完毕，再去消费下一批数据。

package com.wunaiieq.offset;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;


import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;


public class ConsumerHandSyncCommit {
    public static void main(String[] args) {
        //1.创建属性对象
        Properties prop = new Properties();
        //2.设置相关参数
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        //配置消费者组
        prop.put(ConsumerConfig.GROUP_ID_CONFIG,"cghandSyncCommit");
        //设置为非自动提交
        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
        //3.创建消费者对象
        KafkaConsumer<String,String> consumer=
                new KafkaConsumer<String, String>(prop);
        //4.注册消费主题
        consumer.subscribe(Arrays.asList("topicA"));
        //5.消费数据
        while(true){
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofSeconds(1));
            for(ConsumerRecord record:records){
                System.out.println(record.value());
            }
            //6.同步提交offset
            consumer.commitSync();
        }
    }
}

代码示例（异步）

代码上的区别很小，提交方式由consumer.commitSync();改为consumer.commitAsync();

package com.wunaiieq.offset;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;


import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;


public class ConsumerHandASyncCommit {
    public static void main(String[] args) {
        //1.创建属性对象
        Properties prop = new Properties();
        //2.设置相关参数
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        //配置消费者组
        prop.put(ConsumerConfig.GROUP_ID_CONFIG,"cghandAsyncCommit");
        //设置为非自动提交
        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
        //3.创建消费者对象
        KafkaConsumer<String,String> consumer=
                new KafkaConsumer<String, String>(prop);
        //4.注册消费主题
        consumer.subscribe(Arrays.asList("topicA"));
        //5.消费数据
        while(true){
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofSeconds(1));
            for(ConsumerRecord record:records){
                System.out.println(record.value());
            }
            //6.异步提交offset
            consumer.commitAsync();
        }
    }
}

其他说明

  1. 一个消费者允许消费多个主题

http://www.kler.cn/a/468503.html

相关文章：

  • 【论文复现】改进麻雀搜索算法优化冷水机组的最优负载调配问题
  • 遗传学的“正反”之道：探寻生命密码的两把钥匙
  • ArkTs 状态管理装饰器
  • 为什么HTTP请求后面有时带一个sign参数（HTTP请求签名校验）
  • rk3568 上Qt5.12.12迁移问题解决
  • cursor 使用技巧
  • 如何通过本地部署的DIFY辅助学习算法(PS可以辅助帮你学习任何想学习的资料)
  • 探索WebAssembly：前端与后端的新未来
  • unity学习6：unity的3D项目的基本界面和菜单
  • MCP（Model Context Protocol）模型上下文协议 进阶篇3 - 传输
  • 互动为王：开源AI智能名片链动2+1模式商城小程序在社群运营中的深度应用与价值探索
  • 解锁AI Agent潜能：智能时代的信息处理利器2（18/30）
  • ES-深度分页问题
  • LeetCode题练习与总结：随机翻转矩阵--519
  • 使用FDBatchMove的几个问题总结
  • 数据结构：ArrayList与顺序表
  • 每日一学——日志管理工具（Graylog）
  • C++和OpenGL实现3D游戏编程【连载19】——着色器光照初步（平行光和光照贴图）（附源码）
  • 主从复制（Redis的特性）
  • 深入探索 Kubernetes：从基础概念到实战运维
  • 复杂对象的创建与组装 - 建造者模式（Builder Pattern）
  • flutter在windows平台中运行报错
  • BOOST 库在机器视觉中的应用及示例代码分析
  • 【C++】B2099 矩阵交换行
  • 基于 GitHub Actions 的流程自动化实践
  • Apache Traffic存在SQL注入漏洞(CVE-2024-45387)