spring-kafka学习笔记-白红宇

spring-kafka学习笔记

阅读量：7069 次

发布时间：2019-06-28

本文共 2760 字，大约阅读时间需要 9 分钟。

1. 简要说明

是Spring下的一个项目。

官方有详细的和，以下所写的内容也是基于此。没有使用Producer，所以暂时只有Consumer部分的代码，以后会补全。

2. consumer

使用了@KafkaListener注解。

2.1. pom.xml

引入以下依赖


    	
     
      org.springframework.kafka
     	
     
      spring-kafka
     	
     
      1.2.0.RELEASE

2.2. 配置类

@Configuration@EnableKafkapublic 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
    
     
      > kafkaListenerContainerFactory() {		ConcurrentKafkaListenerContainerFactory
      
        factory = new ConcurrentKafkaListenerContainerFactory<>();		factory.setConsumerFactory(consumerFactory());		factory.setConcurrency(3);		factory.getContainerProperties().setPollTimeout(3000);		return factory;	}	@Bean	public ConsumerFactory
       
         consumerFactory() {		return new DefaultKafkaConsumerFactory<>(consumerConfigs());	}	@Bean	public Map
        
          consumerConfigs() {		Map
         
           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
    
      msg) {		// 消费到数据后的处理逻辑		}}

2.4. @KafkaListener和@KafkaListners

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

@KafkaListeners({@KafkaListener(topics="topic1"), @KafkaListener(topics="topic2")})public void listen(ConsumerRecord
    
      msg) {}等同于@KafkaListener(topics="topic1")@KafkaListener(topics="topic2")public void listen(ConsumerRecord
     
       msg) {}