kafka消费者知识整理

旧城等待， 2023-07-24 14:45 141阅读 0赞

说到kafka的消费者，我们要先提到一个概念。  
**消费者组**  
消费者组的意思是：多个消费者可以属于同一个消费者组。投递到主题中的消息只能被消费者组中的一个消费者消费。如果我们要做消息点对点发布，就可以将多个消费者放到一个消费者组中。这样，一条消息只会被一个消费者消费。如果要做消息的发布订阅，就可以将多个消费者分到不同的消费者组中，不同消费者组的消费者都可以消费这条消息。  
通过消费者组，kafka可以实现点对点模型和发布订阅模型的任意转换，非常方便。  
每一个消费者实例一定要属于某一个组，如果我们在实例化某一个消费者的时候，没有指定消费者组ID，此时启动消费者的时候就会报错

Exception in thread "main" org.apache.kafka.common.errors.InvalidGroupIdException: The configured groupId is invalid

关于消费者和消费者组，参考<深入理解kafka>书中的图，说明一下：  
比如，我现在有一个主题，主题中有7个分区。有一个消费者订阅了这个主题，那么这个消费者会消费7个分区中的消息，如图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70]  
这个时候，又有一个消费者订阅了这个主题，同时他们两个属于同一个组。此时，C0和C1会一起消费这7个分区，如图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 1]  
如果消费者属于2个不同的组呢，此时每个组都会订阅所有分区的消息。但是，一条消息  
只会被组中的一个消费者消费。如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 2]  
由以上分析可以得出，一个消费者组的消费者数量是不能多于分区数的，如果多于分区数，就会有消费者没有分区可以消费，如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 3]  
以上逻辑，都是基于默认的分区分配策略进行分析的。可以通过修改消费者端参数partition.assignment.strategy来改变以上策略。  
下面写一个例子，来演示一下如何使用kafka的消费者订阅一个主题，并拉取消息进行处理  
**消费者者API**

//定义消费者的父类
    public class BaseConsumerFast { 
        protected static String brokerList = "192.168.11.81:9092,192.168.11.81:9093,192.168.11.81:9094";
        protected static String groupId = "group.demo";
        protected static String clientId = "consumer.client.id.demo";
        protected static String topic = "topic-partitions";
    
        protected static Properties initConfig(){ 
            Properties properties = new Properties();
            //kafka服务端地址(必填)
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
            //key的反序列化器(必填)
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            //value的反序列化器(必填)
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            //自动提交设置为false,手动提交。默认值为：true
            properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
            //groupId，消费者组标识(必填)
            properties.put(ConsumerConfig.GROUP_ID_CONFIG,groupId);
            //clientId，kafka服务端可以使用kafka-configs.sh脚本，搭配该参数查看消费者的参数设置
            properties.put(ConsumerConfig.CLIENT_ID_CONFIG,clientId);
            return properties;
        }
    }
    //消费者停止消费的控制类
    public class ISRunning { 
        private boolean flag = true;
    
        public boolean get(){ 
            return flag;
        }
    
        public void set(boolean flag){ 
            this.flag = flag;
        }
    }
    //定义消费者
    public class ConsumerFastStart2 extends BaseConsumerFast{ 
        private static ISRunning isRunning = new ISRunning();
        /** * 粗粒度的消息拉取 * @param args */
        public static void main(String[] args) { 
            KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer(initConfig());
            //订阅一个主题
            kafkaConsumer.subscribe(Arrays.asList(topic));
            //获取订阅的所有主题的消息,进行消费
            while (isRunning.get()) { 
            //拉取消息，超时时间为1秒
                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(1000L);
                System.out.println("本次拉取的消息数量:"+consumerRecords.count());
                System.out.println("消息集合是否为空:"+consumerRecords.isEmpty());
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) { 
                    System.out.println("topic:"+consumerRecord.topic()+"---"+"partition:"+consumerRecord.partition()+"---"+"offset:"+consumerRecord.offset());
                    System.out.println("key:"+consumerRecord.key()+"---"+consumerRecord.value());
                    //processing record
                }
            }
        }
    }

可以看到，kafka消费者订阅消息的api是subscribe方法，subscibe方法的入参是一个List集合。一个消费者可以订阅多个主题。如下：

kafkaConsumer.subscribe(Arrays.asList("topic1","topic2"));

订阅方法的重载方法如下：

//以正则表达式方式订阅主题,同时注册了一个rebalance的监听器
    public void subscribe(Pattern pattern, ConsumerRebalanceListener listener)
    //订阅一组主题
    public void subscribe(Collection<String> topics)
    //订阅一组主题,同时注册了一个rebalance的监听器
    public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener)

我上面写的是kafka\_0.11.0.3版本的消费者客户端，不同版本的消费者客户端api是不同的。我们且以0.11.0.3的为例.  
这里我发现，<深入理解kafka>这本书里还介绍了另外一个API  
public void subscribe(Pattern pattern)  
这个方法，是在1.0.0版本开始出现  
对应的API是assign(Collection< TopicPartition > partitions)  
TopicPartition类的定义如下：

public final class TopicPartition implements Serializable { 
        private int hash = 0;
        private final int partition;
        private final String topic;
    
        public TopicPartition(String topic, int partition) { 
            this.partition = partition;
            this.topic = topic;
        }
    
        public int partition() { 
            return this.partition;
        }
    
        public String topic() { 
            return this.topic;
        }

可以看到，里面一共有3个成员变量，hash变量主要和hashcode值相关。我们主要看一下partition和topic这两个变量。topic变量就是指主题，partition变量就是指该主题的某一个分区，比如：我只订阅topic-demo的0号分区

TopicPartition topic = new TopicPartition("topic-demo",0);
    consumer.assign(Arrays.asList(topic));

那如何知道主题有哪些分区呢，kafka提供了partitionsFor( )方法。

//获取当前topic的分区
            List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor(topic);
            for (PartitionInfo partitionInfo : partitionInfos) { 
                TopicPartition tp = new TopicPartition(topic, partitionInfo.partition());
                topicPartitions.add(tp);
            }

PartitionInfo对象的类定义如下：

public class PartitionInfo { 
        private final String topic;
        private final int partition;
        private final Node leader;
        private final Node[] replicas;
        private final Node[] inSyncReplicas;
    
        public PartitionInfo(String topic, int partition, Node leader, Node[] replicas, Node[] inSyncReplicas) { 
            this.topic = topic;
            this.partition = partition;
            this.leader = leader;
            this.replicas = replicas;
            this.inSyncReplicas = inSyncReplicas;
        }

topic，代表分区属于哪个主题  
partition，代表当前是哪个分区  
leader，代表分区的leader节点信息  
replicas，代表分区的AR集合列表  
inSyncReplicas，代表分区的ISR集合列表  
这个地方，1.0.0版本新增了一个

//OSR集合列表
    private final Node[] offlineReplicas;

subscribe和assign方法的区别是：  
subscribe具有消费再均衡的作用，当消费组内的消费者增加或者减少的时候，subscribe可以自动的重新分配分区，以实现消费自动均衡以及故障自动转移。但是assign方法就不具备这个功能。这一点从API上也可以看出来，subscribe有一个可以注册再均衡监听器的API，但是assign方法没有。  
**消息消费**  
消费者从kafka获取消息采取的是拉的方式。

ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(1000L);

poll方法中有一个参数，这个参数是一个long类型的值，单位是毫秒。代表：超时时间。返回值为：消息集合。当超过1秒没有拉取到消息，消费者会立即返回。  
消息集合中的实际消息对象为：

public class ConsumerRecord<K, V> { 
        public static final long NO_TIMESTAMP = -1L;
        public static final int NULL_SIZE = -1;
        public static final int NULL_CHECKSUM = -1;
        private final String topic;
        private final int partition;
        private final long offset;
        private final long timestamp;
        private final TimestampType timestampType;
        private final int serializedKeySize;
        private final int serializedValueSize;
        private final Headers headers;
        private final K key;
        private final V value;
        private volatile Long checksum;
    }

topic：消息来自于哪个主题  
partition：消息来自于哪个主题的哪个分区  
offset：消息的位移  
timestamp：时间戳  
timestampType：时间戳类型。一共有两种，CreateTime和LogAppendTime，分别代表消息创建的时间戳以及消息追加到日志中的时间戳  
serializedKeySize：key序列化后的大小  
serializedValueSize：value序列化后的大小  
headers：消息的头部内容  
key：消息的key  
value：消息的值  
checksum：消息的CRC32的校验值  
以上获取消息的方式是从主题的角度获取所有分区的消息。我们可以将获取消息的粒度缩小为分区级别。我们也可以从消息集中获取某个主题的某个分区的消息。方法为：

List<ConsumerRecord<String, String>> records = consumerRecords.records(new TopicPartition(topic, 0));

records方法还有一个重载方法，records(topic)。按照主题维度获取该主题的所有消息。  
**kafka消费者的关闭**  
kakfa消费者提供了wakeUp方法用来关闭消费者，同时我们也可以自己设置开关进行kafka消费者的关闭。如上面例子中的IsRunning类，就是开关类

**重要的消费者参数**  
1 .fetch.min.bytes  
poll方法，每次返回的数据量。默认为1B，如果kafka服务端没这么多消息，那poll方法将阻塞等待。在生产环境中适当的增大此值，可以提高吞吐量，但是也会造成延时高  
2 .fetch.max.bytes  
与fetch.min.bytes相对，这个参数是指poll方法在kafka服务端所能拉取的最大消息量，默认是50MB。如果单条消息的值大于50MB，poll方法仍将返回，保证消费者可以正常工作  
3 .fetch.max.wait.ms  
poll方法阻塞的最长时间，如果poll方法等了这么久，但是消息还是没有攒够fetch.min.ms的值，此时仍将要返回，防止poll方法无限阻塞下去。默认值是500ms  
4 .max.partition.fetch.bytes  
一次拉取中，一个分区最大能返回的消息量，默认值：1MB  
5 .max.poll.records  
一次拉取中，poll方法最多能返回的消息条数，默认为：500条。如果消息较小，可以适当的增大此值。提升消费速度  
6 .connection.max.idle.ms  
指定多久之后，kafka关闭闲置的链接，默认为：9分钟  
7 .exclude.internal.topics  
kafka的内部主题。\_consumer\_offsets和\_transaction\_state。默认值true的情况下，只有subscribe(Collections)方法可以消费这两个主题的消息。如果将该值设定为false，那subscribe(Collections)和subscribe(Pattern)都可以消费内部主题。  
8 .request.timeout.ms  
这个参数用来设定consumer消费者客户端连接kafka服务端的超时时间，默认值：30秒  
9 .metadata.max.age.ms  
强制更新kafka元数据的时间，默认值：30秒  
10 .reconnect.backoff.ms  
这个参数用来配置尝试连接主机的间隔时间，默认值50(ms)，防止kakfa消费者频繁连接主机，造成资源浪费  
11 .retry.backoff.ms  
这个参数用来配置尝试重新发送失败的请求到指定的主题分区之前的等待时间，避免在某些故障情况下频繁的重复发送，默认值100ms  
12 .isolation.level  
这个参数用来配置消费者的事务隔离级别。可选值有两个，“read\_uncommitted"和"read\_committed”，如果设定为read-committed，kafka消费者将只能读取到LSO的位置。如果设定为read-uncommitted。kafka就可以读到HW的位置  
13 .receive.buffer.bytes  
这个参数用来设定接收消息缓冲区的大小，默认值为：64KB。如果kafka的消费者和服务端跨机房的话，可以适当增大该值。如果设定为-1的话，将会采用操作系统的值  
14 .send.buffer.bytes  
这个参数用来设定发送消息缓冲区的大小，默认值为：128KB。如果设定为-1的话，就会采用操作系统的值  
关于消费者的知识非常多，先写到这里，下一篇介绍位移提交

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70]: /images/20230528/d53d979667cc4f1489c2ce9773ff5c39.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 1]: /images/20230528/73064d849be348858f7988886339b771.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 2]: /images/20230528/3e7d416a403945f8b4f39d0969e82e5e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70 3]: /images/20230528/a7494e3719b041d1866fa1298384f185.png