1. 消费者的结构

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

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

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

1.1 消费者组

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

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

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

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

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

2. 消费者实现

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

参数	解释
bootstrap.servers	集群地址
key.deserializer	key反序列化器
value.deserializer	value反序列化器
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());
            }
        }
    }
}