Kafka高质量专栏请看 石臻臻的杂货铺的Kafka专栏
@KafkaListener(id = "consumer-id",topics = "SHI_TOPIC1",concurrency = "${listen.concurrency:3}",
clientIdPrefix = "myClientId")
属性concurrency
将会从容器中获取listen.concurrency
的值,如果不存在就默认用3
填写:
2020-11-19 14:24:15 c.d.b.k.KafkaListeners 120 [INFO] 线程:Thread[
consumer-id5-1-C-1
,5,main]-groupId:BASE-DEMO consumer-id5 消费
没有填写ID:
2020-11-19 10:41:26 c.d.b.k.KafkaListeners 137 [INFO] 线程:Thread[
org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1
,5,main] consumer-id7
否则会报错
Caused by: java.lang.IllegalStateException: Another endpoint is already registered with id
假如配置文件属性配置了消费组kafka.consumer.group-id=BASE-DEMO
正常情况它是该容器中的默认消费组
但是如果设置了 @KafkaListener(id = "consumer-id7", topics = {"SHI_TOPIC3"})
那么当前消费者的消费组就是consumer-id7
;
当然如果你不想要他作为groupId的话 可以设置属性idIsGroup = false
;那么还是会使用默认的GroupId;
@KafkaListener(id = "consumer-id5",idIsGroup = false,topics = "SHI_TOPIC3",groupId = "groupId-test")
例如上面代码中最终这个消费者的消费组GroupId
是 “groupId-test”
该id属性(如果存在)将用作Kafka消费者group.id属性,并覆盖消费者工厂中的已配置属性(如果存在)您还可以groupId显式设置或将其设置idIsGroup为false,以恢复使用使用者工厂的先前行为group.id。
指定该消费组的消费组名; 关于消费组名的配置可以看看上面的 id 监听器的id
在监听器中调用KafkaUtils.getConsumerGroupId()
可以获得当前的groupId; 可以在日志中打印出来; 可以知道是哪个客户端消费的;
可以同时监听多个
topics = {"SHI_TOPIC3","SHI_TOPIC4"}
可以为监听器配置明确的主题和分区(以及可选的初始偏移量)
@KafkaListener(id = "thing2", topicPartitions =
{
@TopicPartition(topic = "topic1", partitions = {
"0", "1" }),
@TopicPartition(topic = "topic2", partitions = "0",
partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
})
public void listen(ConsumerRecord<?, ?> record) {
...
}
上面例子意思是 监听topic1
的0,1分区;监听topic2
的第0分区,并且第1分区从offset为100的开始消费;
实现KafkaListenerErrorHandler
; 然后做一些异常处理;
@Component
public class KafkaDefaultListenerErrorHandler implements KafkaListenerErrorHandler {
@Override
public Object handleError(Message<?> message, ListenerExecutionFailedException exception) {
return null;
}
@Override
public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
//do someting
return null;
}
}
调用的时候 填写beanName;例如errorHandler="kafkaDefaultListenerErrorHandler"
指定生成监听器的工厂类;
例如我写一个 批量消费的工厂类
/** * 监听器工厂 批量消费 * @return */
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> batchFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(kafkaConsumerFactory());
//设置为批量消费,每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
factory.setBatchListener(true);
return factory;
}
使用containerFactory = "batchFactory"
会覆盖消费者工厂的
kafka.consumer.client-id
属性; 最为前缀后面接-n
n是数字
会覆盖消费者工厂中的concurrency ,这里的并发数就是多线程消费; 比如说单机情况下,你设置了3; 相当于就是启动了3个客户端来分配消费分区;分布式情况 总线程数=concurrency*机器数量; 并不是设置越多越好,具体如何设置请看 属性concurrency的作用及配置(RoundRobinAssignor 、RangeAssignor)
/** * 监听器工厂 * @return */
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> concurrencyFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(kafkaConsumerFactory());
factory.setConcurrency(6);
return factory;
}
@KafkaListener(id = "consumer-id5",idIsGroup = false,topics = "SHI_TOPIC3", containerFactory = "concurrencyFactory",concurrency = "1)
虽然使用的工厂是concurrencyFactory
(concurrency配置了6); 但是他最终生成的监听器数量 是1;
kafka中的属性看org.apache.kafka.clients.consumer.ConsumerConfig
;
同名的都可以修改掉;
用法
@KafkaListener(id = "consumer-id5",idIsGroup = false,topics = "SHI_TOPIC3", containerFactory = "concurrencyFactory",concurrency = "1"
, clientIdPrefix = "myClientId5",groupId = "groupId-test",
properties = {
"enable.auto.commit:false","max.poll.interval.ms:6000" },errorHandler="kafkaDefaultListenerErrorHandler")
@Autowired
private KafkaListenerEndpointRegistry registry;
//.... 获取所有注册的监听器
registry.getAllListenerContainers();
当您将Spring Boot与验证启动器一起使用时,将LocalValidatorFactoryBean自动配置:如下
@Configuration
@EnableKafka
public class Config implements KafkaListenerConfigurer {
@Autowired
private LocalValidatorFactoryBean validator;
...
@Override
public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
registrar.setValidator(this.validator);
}
}
使用
@KafkaListener(id="validated", topics = "annotated35", errorHandler = "validationErrorHandler",
containerFactory = "kafkaJsonListenerContainerFactory")
public void validatedListener(@Payload @Valid ValidatedClass val) {
...
}
@Bean
public KafkaListenerErrorHandler validationErrorHandler() {
return (m, e) -> {
...
};
}
TODO…
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
发布者:全栈程序员栈长,转载请注明出处:https://javaforall.cn/181902.html原文链接:https://javaforall.cn