springboot整合kafka实现发布订阅，kafka学习五

梦里梦外; 2021-06-22 15:38 883阅读 0赞

## 1.引入依赖 ##

<dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
         </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
             <optional>true</optional>
         </dependency>

## 2.yml配置 ##

spring:
      kafka:              # 指定kafka 代理地址，可以多个
        bootstrap-servers: 192.168.211.137:9092,192.168.211.139:9092,192.168.211.140:9092
        template:    # 指定默认topic id
          default-topic: producer
        listener:   # 指定listener 容器中的线程数，用于提高并发量
          concurrency: 5
        consumer:
          group-id: myGroup # 指定默认消费者group id
          client-id: 200
          max-poll-records: 200
          auto-offset-reset: earliest # 最早未被消费的offset
        producer:
          batch-size: 1000 # 每次批量发送消息的数量
          retries: 3
          client-id: 200

## 3.代码示例 ##

1.  生产者

package com.wyu.tt06kafkademo.demo;
    
    import com.fasterxml.jackson.core.JsonProcessingException;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    
    @Slf4j
    @Component
    public class KafkaProducer { 
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        @Autowired
        private ObjectMapper objectMapper;
    
        public void send(String topic, Object body) { 
            Message<String> message = new Message<>();
            message.setId(System.currentTimeMillis());
            message.setMessage(body.toString());
            message.setTime(new Date());
            String content = null;
            try { 
                content = objectMapper.writeValueAsString(message);
            } catch (JsonProcessingException e) { 
                e.printStackTrace();
            }
            kafkaTemplate.send(topic, content);
            log.info("send {} to {} success!", message, topic);
            System.out.println("send "+ message +" to "+ topic +" success!");
        }
    }

1.  消费者

package com.wyu.tt06kafkademo.demo;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    import java.util.List;
    import java.util.Optional;
    
    @Slf4j
    @Component
    public class KafkaConsumer { 
    
        /**
         * 有消息就读取,读取消息topic，offset，key，value等信息
         */
        @KafkaListener(topics = { "kafka1"})
        public void listen(ConsumerRecord<?, ?> record) { 
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) { 
                Object message = kafkaMessage.get();
                log.info("详细消息读取-------------------->");
                log.info("message:{} + record:{}", message, record);
            }
        }
    
        /**
         * 有消息就读取,批量读取消息
         */
        @KafkaListener(topics = "kafka1")
        public void onMessage(List<String> crs) { 
            for(String str : crs){ 
                log.info("批量读取-------------------->");
                log.info("kafka1:" + str);
            }
        }
    
        /**
         * 有消息就读取message
         */
        @KafkaListener(topics = { "kafka1"})
        public void receiveMessage(String message){ 
            log.info("读取message-------------------->");
            log.info("kafka1:" + message);
        }
    }

1.  controller

package com.wyu.tt06kafkademo.controller;
    
    import com.wyu.tt06kafkademo.demo.KafkaProducer;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    
    @Slf4j
    @RestController
    public class KafkaController { 
        @Autowired
        private KafkaProducer kafkaProducer;
    
        @GetMapping("/kafka/{topic}")
        public String send(@PathVariable("topic") String topic, @RequestParam String message) { 
            kafkaProducer.send(topic, message);
            return "success";
        }
    }