前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >@KafkaListener和@KafkaListeners的使用

@KafkaListener和@KafkaListeners的使用

作者头像
小勇DW3
发布2019-11-14 12:47:43
5.4K0
发布2019-11-14 12:47:43
举报
文章被收录于专栏:小勇DW3

2. consumer

使用了@KafkaListener注解。

2.1. pom.xml

引入以下依赖

代码语言:javascript
复制
<dependency>
	<groupId>org.springframework.kafka</groupId>
	<artifactId>spring-kafka</artifactId>
	<version>1.2.0.RELEASE</version>
</dependency>
2.2. 配置类
代码语言:javascript
复制
@Configuration
@EnableKafka
public class KafkaConfig {
	
	@Value("${kafka.bootstrap.servers}")
	private String kafkaBootstrapServers;
	@Value("${session.timeout.ms}")
	private Integer sessionTimeoutMs;
	@Value("${enable.auto.commit}")
	private boolean enableAutoCommit;
	@Value("${auto.commit.interval.ms}")
	private Integer autoCommitIntervalMs;
	@Value("${auto.offset.reset}")
	private String autoOffsetReset;
	@Value("${group.id}")
	private String groupId;
	
	@Bean
	KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
		ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(consumerFactory());
		factory.setConcurrency(3);
		factory.getContainerProperties().setPollTimeout(3000);
		return factory;
	}

	@Bean
	public ConsumerFactory<Integer, String> consumerFactory() {
		return new DefaultKafkaConsumerFactory<>(consumerConfigs());
	}

	@Bean
	public Map<String, Object> consumerConfigs() {
		Map<String, Object> props = new HashMap<>();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers);
		props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs);
		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		return props;
	}

	@Bean
	public KafkaConsumer consumer() {
		return new KafkaConsumer();
	}
	
	
}
2.3. 消费处理
代码语言:javascript
复制
public class KafkaConsumer {
 
	@KafkaListeners({@KafkaListener(topics="topic1"), @KafkaListener(topics="topic2")})
	public void listen(ConsumerRecord<Integer, String> msg) {
		// 消费到数据后的处理逻辑	
	}
}
2.4. @KafkaListener和@KafkaListners

@KafkaListeners是@KafkaListener的Container Annotation,这也是jdk8的新特性之一,注解可以重复标注。

代码语言:javascript
复制
@KafkaListeners({@KafkaListener(topics="topic1"), @KafkaListener(topics="topic2")})
public void listen(ConsumerRecord<Integer, String> msg) {}

等同于

@KafkaListener(topics="topic1")
@KafkaListener(topics="topic2")
public void listen(ConsumerRecord<Integer, String> msg) {}
2.5. @KafkaListener使用小结

在注解上可以方便地进行各种配置,但是如果要消费的topic个数不定,用@KafkaListener就很难优雅解决。

注解要求必须在compile-time就能确定值,可以移步stackOverFlow查看更加详细的解释。

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2019-11-13 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 2. consumer
    • 2.1. pom.xml
      • 2.2. 配置类
        • 2.3. 消费处理
          • 2.4. @KafkaListener和@KafkaListners
            • 2.5. @KafkaListener使用小结
            领券
            问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档