【大数据学习 | kafka】简述kafka的消费者consumer

news/2024/11/5 5:56:42 标签: mysql, kafka, flume, storm, zookeeper

1. 消费者的结构

能够在kafka中拉取数据进行消费的组件或者程序都叫做消费者。

这里面要涉及到一个动作叫做拉取。

首先我们要知道kafka这个消息队列主要的功能就是起到缓冲的作用，比如flume采集数据然后交给spark或者flink进行计算分析，但是flume采用的就是消息的push方式，这个方式不能够保证推送的数据消费者端一定会消费完毕，会出现数据的反压问题，这个问题很难解决，所以才出现了消息队列kafka，它可以起到一个缓冲的作用，生产者部分将数据直接全部推送到kafka，然后消费者从其中拉取数据，这边如果也采用推送的方式，那么也就在计算端会出现反压问题，所以kafka的消费者一般都是采用拉的方式pull，并不是push

1.1 消费者组

在一个topic中存在多个分区，可以分摊压力实现负载均衡，那么整体topic中的数据会很多，如果消费者只有一个的话很难全部消费其中的数据，压力也会集中在一个消费者中，并且在大数据行业中几乎所有的计算架构都是分布式的集群模式，那么这个集群模式中，计算的节点也会存在多个，这些节点都是可以从kafka中拉取数据的，所有消费者不可能只有一个，一般情况下都会有多个消费者。

正因为topic存在多个分区，每个分区中的数据是独立的，那么消费者最好也是一个一个和分区进行一一对应的，所以有几个分区应该对应存在几个消费者是最好的。

这个和分蛋糕是一样的，一个蛋糕分成几块，那么有几个人吃，应该是对应关系的

消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。

消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。

2. 消费者实现

在实现消费者的时候我们需要知道几个消费者的配置重要参数

参数解释
bootstrap.servers集群地址
key.deserializerkey反序列化器
value.deserializervalue反序列化器
group.id消费者组id

首先创建消费者对象

消费者对象订阅相应的topic然后拉取其中的数据进行消费

整体代码如下

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.Iterator;
import java.util.List;
import java.util.Properties;

public class Consumer1 {
    public static void main(String[] args) {
        Properties pro = new Properties();
        pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"nn1:9092");
        pro.put(ConsumerConfig.GROUP_ID_CONFIG,"hainiu_group");
        //设定组id
        pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //设定key的反序列化器
        pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //设定value的反序列化器
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(pro);
        List<String> topics = Arrays.asList("topic_a","topic_b");
        //一个消费者可以消费多个分区的数据
        consumer.subscribe(topics);
        //订阅这个topic
        while (true){
            //死循环要一直消费数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            //间隔一秒钟消费一次数据，拉取一批数据过来
            Iterator<ConsumerRecord<String, String>> it = records.iterator();
            while(it.hasNext()){
                ConsumerRecord<String, String> record = it.next();
                System.out.println(record.topic()+"->"+record.partition()+"->"+ record.offset()+"->"+record.key()+"->"+record.value());
            }
        }
    }
}
[hexuan@hadoop106 datas]$ kafka-console-producer.sh --bootstrap-server hadoop106:9092 --topic topic_b

>>1
>2
>3
>4
>5
>

3. 消费者与分区之间的对应关系

一个消费者组中的消费者和分区是一一对应的关系，一个分区应该对应一个消费者，但是如果消费者多了，那么有的消费者就没有分区消费，如果消费者少了那么会出现一个消费者消费多个分区的情况。

# 首先创建topic_c 用于测试分区和消费者的对应关系
kafka-topics.sh --bootstrap-server hadoop106:9092 --create --topic topic_c --partitions 3 --replication-factor 2
# 启动两个消费者 刚才我们写的消费者main方法运行两次
# 然后分别在不同的分区使用生产者发送数据，看数据在消费者中的打印情况

首先选择任务可以并行执行

选择任务修改配置

我们可以看到允许多实例并行执行

启动两次，这个时候我们就有了两个消费者实例

生产者线程:分别向三个分区中发送1 2 3元素

package com.hainiu.kafka.consumer;

/**
 * ClassName : test3_producer
 * Package : com.hainiu.kafka.consumer
 * Description
 *
 * @Author HeXua
 * @Create 2024/11/3 23:40
 * Version 1.0
 */

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class test3_producer {
    public static void main(String[] args) {
        Properties pro = new Properties();
        pro.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop106:9092");
        pro.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        pro.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(pro);
        ProducerRecord<String, String> record1 = new ProducerRecord<>("topic_d", 0,null,"1");
        ProducerRecord<String, String> record2 = new ProducerRecord<>("topic_d", 1,null,"2");
        ProducerRecord<String, String> record3 = new ProducerRecord<>("topic_d", 2,null,"3");
        producer.send(record1);
        producer.send(record2);
//        producer.send(record3);
        producer.close();
    }
}

可以看到有的消费者消费了两个分区的数据

如果启动三个消费者会发现每个人消费一个分区的数据

如果启动四个消费者

我们发现有一个消费者没有数据

3. 1 消费多topic的数据

不同组消费不同的topic或者一个组可以消费多个topic都是可以的

3.2 多个组消费一个topic

同一个topic可以由多个消费者组进行消费数据，并且相互之间是没有任何影响的

修改同一份代码的组标识不同。启动两个实例查看里面的消费信息

   pro.put(ConsumerConfig.GROUP_ID_CONFIG,"hainiu_group1");
   pro.put(ConsumerConfig.GROUP_ID_CONFIG,"hainiu_group2");
   //分别修改消费者组的id不同
package com.hainiu.kafka;

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.Iterator;
import java.util.List;
import java.util.Properties;

public class Consumer1 {
    public static void main(String[] args) {
        Properties pro = new Properties();
        pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"nn1:9092");
        pro.put(ConsumerConfig.GROUP_ID_CONFIG,"hainiu_group");
        pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(pro);
        List<String> topics = Arrays.asList("topic_c");
        //订阅多个topic的数据变化
        consumer.subscribe(topics);

        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            Iterator<ConsumerRecord<String, String>> it = records.iterator();
            while(it.hasNext()){
                ConsumerRecord<String, String> record = it.next();
                System.out.println(record.topic()+"->"+record.partition()+"->"+ record.offset()+"->"+record.key()+"->"+record.value());
            }
        }
    }
}


http://www.niftyadmin.cn/n/5739030.html

相关文章

JavaFx -- chapter06(UDPSocket)

JavaFx -- chapter06(UDPSocket)

chapter06(UDPSocket) UPD的特点 UDP有独立的套接字&#xff08;IP PORT&#xff09;&#xff0c;与TCP使用相同端口号不会冲突。UDP在使用前不需要进行连接&#xff0c;没有流的概念。UDP通信类似于邮件通信&#xff1a;不需要实时连接&#xff0c;只需要目的地址。UDP通信…
阅读更多...
OceanBase 安装使用详细说明

OceanBase 安装使用详细说明

OceanBase 安装使用详细说明 一、系统环境要求二、安装OceanBase环境方案一：在线下载并安装all-in-one安装包方案二：离线安装all-in-one安装包安装前的准备工作三、配置OceanBase集群编辑配置文件部署和启动集群连接到集群集群状态和管理四、创建业务租户和数据库创建用户并赋…
阅读更多...
Python 单元测试中的 Mocking 与 Stubbing：提高测试效率的关键技术

Python 单元测试中的 Mocking 与 Stubbing：提高测试效率的关键技术

在软件开发过程中&#xff0c;单元测试是确保代码质量的重要环节。为了实现高效的单元测试&#xff0c;我们常常需要隔离待测试的代码与其外部依赖。这时候&#xff0c;Mocking&#xff08;模拟&#xff09;和 Stubbing&#xff08;桩&#xff09;技术就显得尤为重要。这两种技…
阅读更多...
6.机器学习--PCA主成分分析（降维）

6.机器学习--PCA主成分分析（降维）

目录 1.问题的引入 为什么要降维&#xff1f; 降维的好处 降维的本质 2.降维的主要方法&#xff1a; 2.1 特征选择 2.2 特征抽取 3.主成分分析&#xff08;PCA&#xff09;推导 3.1.向量的表示及基变换 3.2.协方差矩阵及优化目标 3.3.算法及实例 3.4.实例 3.5.代…
阅读更多...
Oceanbase学习之一迁移mysql数据到oceanbase

Oceanbase学习之一迁移mysql数据到oceanbase

一、数据库环境 #mysql环境 root192.168.150.162 20:28: [(none)]> select version(); ---------- | version() | ---------- | 8.0.26 | ---------- 1 row in set (0.00 sec) root192.168.150.162 20:28: [(none)]> show variables like ‘%char%’; ---…
阅读更多...
提交linux kernel patch流程

提交linux kernel patch流程

缩进 git commit 要求自动换行(按照社区具体已要求&#xff0c;内核要求75字符) git config --global core.editor "vim" ~/.vimrc 中写入 autocmd FileType gitcommit setlocal textwidth75 本地提交commit git commit -s -s会自动在结尾加上Signed-off-by&…
阅读更多...
论负载均衡技术在Web系统中的应用

论负载均衡技术在Web系统中的应用

论负载均衡技术在Web系统中的应用 引言 在当今高并发的互联网环境中&#xff0c;负载均衡技术已经成为提升Web系统性能不可或缺的一环。通过有效地将请求分发到多个服务器上&#xff0c;负载均衡不仅能够提高系统的响应速度和处理能力&#xff0c;还能增强系统的可扩展性和稳…
阅读更多...
w~自动驾驶合集11

w~自动驾驶合集11

我自己的原文哦~ https://blog.51cto.com/whaosoft/12329152 #特斯拉的“纯视觉”路线 , 也许不是最好的 BEVTransformer占用网络技术路线的大热&#xff0c;再次将激光雷达推向风口浪尖。 激光雷达该不该被抛弃&#xff1f; 对车企来说&#xff0c;这是一个艰难的抉择&am…
阅读更多...
最新文章

Copyright @ 2022~2023