spring整合kafka

桃扇骨 2022-10-12 01:02 155阅读 0赞

1、导入pom依赖，kafka的客户端版本要与kafka的服务端版本一致

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

2、注入KafkaTemplate模板

@Configuration
    @EnableKafka
    public class KafkaConfig {
    
        private final  static  String  CONSUMER_GROUP_ID="yd-group";
    
        public  final static   String TOPIC_NAME="yd-kf-topic";
    
        @Bean
       public ConcurrentKafkaListenerContainerFactory<String, String>  kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            return factory;
        }
    
        /**
         * 消费工厂
         * @return
         */
        @Bean
        public ConsumerFactory<String, String> consumerFactory() {
            Map<String, Object> props = new HashMap<>(8);
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "PLAINTEXT://192.168.81.200:9092");
            props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_ID);
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
            props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
            props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            return new DefaultKafkaConsumerFactory<>(props);
        }
    
        /**
         * 生产工厂
         * @return
         */
        @Bean
        public ProducerFactory<String, String> producerFactory() {
            Map<String, Object> props = new HashMap<>(8);
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "PLAINTEXT://192.168.81.200:9092");
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            props.put(ProducerConfig.RETRIES_CONFIG, 0);
            props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
            props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
            props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            return new DefaultKafkaProducerFactory<>(props);
        }
    
        /**
         * kafka模板
         * @return
         */
        @Bean("kafkaTemplate")
        public KafkaTemplate<String, String> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }
    }

3、消息生产、消息消费监听

消息发送

@Slf4j
    @Service
    public class KafkaProducer {
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        public  String  sendSyncMessage(String key,String  msg){
            String s;
            try {
                ListenableFuture<SendResult<String, String>> tagA = kafkaTemplate.send(KafkaConfig.TOPIC_NAME, key, msg);
                s = tagA.get().toString();
                log.info("生产kafka消息 {}",s);
                return  s;
            } catch (InterruptedException|ExecutionException e) {
                e.printStackTrace();
                s=e.getMessage();
                log.error("sendSyncMessage-->发送消息异常{}",e.getMessage());
            }
            return s;
        }
    }

监听消息消费

@Slf4j
    @Component
    public class CustomKafkaListener  /**implements MessageListener<String,String>*/ {
    
        @KafkaListener(topics = {KafkaConfig.TOPIC_NAME},id = KafkaConfig.TOPIC_NAME)
        public   void   onMessage1(String  msg){
            log.info("onMessage1消费kafka消息 {} ",msg);
        }
        
    }

4、测试发送

@RestController
    public class KafkaSendController {
        
        @Autowired
        private KafkaProducer kafkaProducer;
    
        @GetMapping("/kafka/sendMsg")
        public  String  sendMsg(String  key,String msg){
            return kafkaProducer.sendSyncMessage(key,msg);
        }
    
    }