ubuntu 建设网站19
kafka 消息订阅与消息消费几种方式
本文主要内容
-
消费者订阅几种方式
-
订阅多个主题
-
按正则表达式订阅
-
-
消息消费几种方式
-
按分区消费
-
按主题消费
-
不区分
-
“笔者建议一开始学习Kafka最好不要用SpringBoot 集成方式,因为SpringBoot推崇用注解方式，比如
@KafkaListener等，就可以直接消费，这样不能直接接触kafka-client一些api, 且SpringBoot 给我们提供了很多默认配置，我们几乎零配置也可以使用，实际上kafka很多配置很重要的，不容忽视。
消费者订阅几种方式
KafkaConsumer 给我们提供了几种订阅消息方式，我们可以订阅多个消息。示例代码如下
kafkaConsumer.subscribe(Arrays.asList("topicA","topicB"));kafkaConsumer.subscribe(Pattern.compile("topic-*"));kafkaConsumer.assign(Arrays.asList(new TopicPartition("topicA",0)));
订阅多个主题
void subscribe(Collection<String> topics) 对应上面第一行代码,这是最常见的订阅方式
按正则表达式订阅
void subscribe(Pattern pattern) 符合正则的主题都会被消费
“有人创建了新的主题，并且与正则匹配，消费者也可以消费到
这种方式需要能对多种消息处理，对于一些能通用处理，不感知具体业务数据的场景比较合适。比如B系统需要同步A系统数据，我们按正则订阅，当A系统有新的数据需要同步，这是只需要A发满足条件正则的消息，B系统无需任何改动。
订阅指定分区
void assign(Collection<TopicPartition> partitions);正常业务不会使用，如果订阅的分区不存在，会报错。一些特殊场景，比如需要精确控制消费者消费消息，自定义分区分配策略时 可能会用到assign 方法
消息消费
“kafka 采用客户端
拉取模式进行消息消费
poll() 返回所订阅的主题上一组消息ConsumerRecords ,我们可以对消息进行按主题、按分区进行处理，当然可以统一处理，不分主题和分区
ConsumerRecords<String,String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
不区分主题、分区
for(ConsumerRecord<String,String> record : records){// 处理消息
}
按partition 处理
Set<TopicPartition> topicPartitions = records.partitions();
for(TopicPartition topicPartition : topicPartitions){List<ConsumerRecord<String, String>> tpRecords = records.records(topicPartition);
}
按主题
Iterable<ConsumerRecord<String,String>> iter = records.records(topic);
思考
kafka 给我们提供了灵活的消息订阅以及消息消费方式，我们需要根据实际业务场景选择。无论哪种场景都离不开 主题、分区 ，最主要的是分区，当我们选择了某种订阅方式如果主题、分区 发生了变化 ，消息还能正常消费吗
“选择了按正则订阅消息方式， 后面创建了新的主题，该消息能被正常消费吗
“选择了指定分区订阅， 如果后面扩容了新的分区，新分区消息能消费吗？
List<PartitionInfo> partitionsFor(String topic) 能获取分区情况，如果需要按分区订阅，该方法一定用的上
“按分区维度消费消息，对于手动提交消息位移场景非常有用
“按主题分类处理消息也很常见，因为不同主题消息格式可能是不一样的，根据主题区分，很容易将不同的消息分类处理。