@KafkaListener和@KafkaListeners的使用

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

@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();
	}
	
	
}

public class KafkaConsumer {
 
	@KafkaListeners({@KafkaListener(topics="topic1"), @KafkaListener(topics="topic2")})
	public void listen(ConsumerRecord<Integer, String> msg) {
		// 消费到数据后的处理逻辑	
	}
}

@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) {}