@KafkaListener和@KafkaListeners的使用

Posted 2020-11-23 gxyandwmm

2. consumer

使用了@KafkaListener注解。

2.1. pom.xml

引入以下依赖

<dependency>
	<groupId>org.springframework.kafka</groupId>
	<artifactId>spring-kafka</artifactId>
	<version>1.2.0.RELEASE</version>
</dependency>

2.2. 配置类

@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. 消费处理

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的新特性之一，注解可以重复标注。

@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查看更加详细的解释。