在spring中创建kafak生产者

r囧r小猫 2022-10-27 01:36 97阅读 0赞

# 在spring中创建kafak生产者 #

## 1. Spring中添加kafka的依赖 ##

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

## 2. 配置文件中添加kafka的连接信息 ##

server.port=8089
    #指定项目名
    server.context-path=/gmall
    
    #============== kafka ===================
    # 指定kafka 代理地址，可以多个
    spring.kafka.bootstrap-servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
    
    # 指定消息key和消息体的编解码方式
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

## 3. 编写Controloer方法 ##

package com.atguigu.gmalllogger.controller;
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.gmall.common.constansts.GmallConstants;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.log4j.Logger;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    @Slf4j  //自动在容器中创建一个记录当前类信息的Logger，等价于 Logger logger = Logger.getLogger(MainController.class);
    public class MainController { 
    
        @Autowired
        private KafkaTemplate<String,String> kafkaTemplate;
    
        //编写处理逻辑，处理上传的日志
        @RequestMapping("/log")
        public String handleLog(String logString){ 
    
            //System.out.println(logString);
    
            //为logString添加一个时间戳
            JSONObject jsonObject = JSON.parseObject(logString);
            jsonObject.put("ts",System.currentTimeMillis());
    
            //将加完时间戳的数据，转为JsonString
            String jsonString = jsonObject.toJSONString();
    
            //记录日志
            log.info(jsonString);
    
    
            //判断日志类型，将json字符串写入到不同kafka主题
            //typs：startup，type：event
            if ("startup".equals(jsonObject.getString("type"))){ 
                kafkaTemplate.send(GmallConstants.KAFKA_TOPIC_STARTUP,jsonString);
            }else { 
                kafkaTemplate.send(GmallConstants.KAFKA_TOPIC_EVENT,jsonString);
            }
    
            return "success";
    
        }
    }