Spring使用Kafka实现生产者消费者

短命女 2022-05-27 06:47 237阅读 0赞

一.maven引入依赖

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

2.application.properties配置

#========================kafka=====================================
    #kafka相关配置
    spring.kafka.bootstrap-servers=127.0.0.1:9092
    #设置一个默认组
    spring.kafka.consumer.group-id=defaultGroup
    #消费者 key-value反序列化 
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    #生产者 key-value序列化 
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
    #producer将试图批处理消息记录，以减少请求次数。这项配置控制默认的批量处理消息字节数。不会试图处理大于这个字节数的消息字节数。
    spring.kafka.producer.batch-size=65536
    #可以用来缓存数据的内存大小如果数据产生速度大于向broker发送的速度，producer会阻塞或者抛出异常，以“block.on.buffer.full”来表明。
    #这项设置将和producer能够使用的总内存相关，但并不是一个硬性的限制，因为不是producer使用的所有内存都是用于缓存。
    #一些额外的内存会用于压缩（如果引入压缩机制），同样还有一些用于维护请求。
    spring.kafka.producer.buffer-memory=524288

3.生产者

import java.time.LocalDateTime;
    import java.time.format.DateTimeFormatter;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;
    /** * * ClassName: Producer * @Description:kafka生产者 * @date 2018年4月8日 */
    @Component
    public class Producer { 
           /** * 这个类包装了个生产者，来提供方便的发送数据到kafka的topic里面。 * 同步和异步的方法都有，异步方法返回一个Future */
              @Autowired
            private KafkaTemplate<String,Object> kafkaTemplate;
    
            /** * 发送消息到kafka,主题为sbKafka */
            public void sendTest(String str){
                kafkaTemplate.send("sbKafka","北京时间: "+LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd hh:mm:ss")));
            }
    }

4.消费者

import java.util.Arrays;
    import java.util.HashMap;
    import java.util.HashSet;
    import java.util.Map;
    import java.util.Set;
    
    import lombok.extern.slf4j.Slf4j;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.log4j.Logger;
    import org.apache.spark.streaming.api.java.JavaInputDStream;
    import org.apache.spark.streaming.api.java.JavaStreamingContext;
    import org.apache.spark.streaming.kafka010.ConsumerStrategies;
    import org.apache.spark.streaming.kafka010.KafkaUtils;
    import org.apache.spark.streaming.kafka010.LocationStrategies;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    /** * * ClassName: Consumer * @Description: 消费者 * @date 2018年4月8日 */
    
    @Component
    @Slf4j
    public class Consumer {
        @Autowired
        private JavaStreamingContext jssc;
        private Logger log = Logger.getLogger(Consumer.class);
         /** * 监听demo,sbKafka主题,有消息就读取 * @param message */
        @KafkaListener(topics = {
       "demo","sbKafka"})
        public void consumer(String message){
            log.info("message======== "+message);
            System.err.println(message);
        }
    }

5.向kafka发送消息

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    public class KafkaDemo { 
        @Autowired
        private Producer producer;
        @RequestMapping("/kafka")
        public String testKafka(String str) {
            producer.sendTest(str);
            return "ok";
        }
    }

6.当访问 /kafka 路径时,控制台打印消息。