springboot整合logback，自定义发送日志到kafka

刺骨的言语ヽ痛彻心扉 2023-06-24 08:13 7阅读 0赞

**项目动态获取kafka地址，还有对kafka是否在线ping**

这个是通过获取log的日志输出，然后发送到kafka里。部署了一套ELK，logstash负责监听kafka，有消息直接写入到es里

首先项目的maven依赖s

<dependency>
                <groupId>com.github.danielwegener</groupId>
                <artifactId>logback-kafka-appender</artifactId>
                <version>0.1.0</version>
            </dependency>
            <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-core</artifactId>
                <version>1.1.11</version>
            </dependency>
            <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-classic</artifactId>
                <version>1.1.11</version>
            </dependency>
            <dependency>
                <groupId>net.logstash.logback</groupId>
                <artifactId>logstash-logback-encoder</artifactId>
                <version>4.11</version>
            </dependency>

**三个ping相关的类**

Constants.java

/**
     * @ClassName Constants
     * @Author dlh
     * @Date 2019/11/13 18:02
     * @Version 1.0
     **/
    public class Constants {
        /***
         *删除状态
         */
        public static enum DeleteStatus{
            NORMAL("0","NORMAL","正常"),DELETE("1","DELETE","删除");
            private DeleteStatus(String value,String name,String desc){
                this.value=value;
                this.name=name;
                this.desc=desc;
            }
            private final String value;
            private final String name;
            private final String desc;
    
            public String getValue() {
                return value;
            }
    
            public String getName() {
                return name;
            }
    
            public String getDesc() {
                return desc;
            }
        }
    
        /***
         * Result code
         */
        public static enum ResultCode {
            /** 成功 */
            SUCCESS("200", "成功"),
            NULL_DATA("205","无数据"),
            /** 没有登录 */
            NOT_LOGIN("400", "没有登录"),
    
            /** 发生异常 */
            EXCEPTION("401", "发生异常"),
    
            /** 系统错误 */
            SYS_ERROR("402", "系统错误"),
    
            /** 参数错误 */
            PARAMS_ERROR("403", "参数错误 "),
    
            /** 不支持或已经废弃 */
            NOT_SUPPORTED("410", "不支持或已经废弃"),
    
            /** AuthCode错误 */
            INVALID_AUTHCODE("444", "无效的AuthCode"),
    
            /** 太频繁的调用 */
            TOO_FREQUENT("445", "太频繁的调用"),
    
            /** 未知的错误 */
            UNKNOWN_ERROR("499", "未知错误");
    
    
            private ResultCode(String value, String msg){
                this.val = value;
                this.msg = msg;
            }
    
            public String val() {
                return val;
            }
    
            public String msg() {
                return msg;
            }
    
            private String val;
            private String msg;
            }
    }

JsonResult.java

/**
     * @ClassName JsonResult
     * @Author dlh
     * @Date 2019/11/13 18:01
     * @Version 1.0
     **/
    
    public class JsonResult {
        /**
         * 结果编码
         */
        private String code;
        /**
         * 结果描述
         */
        private String message;
        private Object data;
    
        public JsonResult() {
            this.setCode(Constants.ResultCode.SUCCESS);
            this.setMessage(Constants.ResultCode.SUCCESS.msg());
    
        }
    
        public JsonResult(Constants.ResultCode code) {
            this.setCode(code);
            this.setMessage(code.msg());
        }
    
        public JsonResult(Constants.ResultCode code, String message) {
            this.setCode(code);
            this.setMessage(message);
        }
    
        public JsonResult(Constants.ResultCode code, String message, Object data) {
            this.setCode(code);
            this.setMessage(message);
            this.setData(data);
        }
    
        public String getCode() {
            return code;
        }
        public void setCode(Constants.ResultCode code) {
            this.code = code.val();
            this.message = code.msg();
        }
        public String getMessage() {
            return message;
        }
        public void setMessage(String message) {
            this.message = message;
        }
    
        public Object getData() {
            return data;
        }
    
        public void setData(Object data) {
            this.data = data;
        }
    
        @Override
        public String toString() {
            return "JsonResult{" +
                    "code='" + code + '\'' +
                    ", message='" + message + '\'' +
                    ", data=" + data +
                    '}';
        }
    }

PTUtil.java

import java.io.IOException;
    import java.net.InetAddress;
    import java.net.InetSocketAddress;
    import java.net.Socket;
    import java.net.UnknownHostException;
    
    /**
     * @ClassName PTUtil
     * @Author dlh
     * @Date 2019/11/13 18:01
     * @Version 1.0
     **/
    
    public class PTUtil {
        /***
         *  ping操作
         * @param hostname
         * @param timeout in milliseconds
         * @return
         */
        public static JsonResult pingResult(String hostname,Integer timeout){
            JsonResult jsonResult = new JsonResult();
            try {
                InetAddress address = InetAddress.getByName(hostname);
                boolean flag = address.isReachable(timeout);
                if(flag){
                    jsonResult.setMessage("ping结果:the address is reachable.");
                }else{
                    jsonResult.setCode(Constants.ResultCode.EXCEPTION);
                    jsonResult.setMessage("ping结果:the address is unreachable.");
                }
            } catch (UnknownHostException e) {
                jsonResult.setCode(Constants.ResultCode.EXCEPTION);
                jsonResult.setMessage("ping结果:UnknownHostException:"+e.getMessage());
            } catch (IOException e) {
                jsonResult.setCode(Constants.ResultCode.EXCEPTION);
                jsonResult.setMessage("ping结果:IOException:"+e.getMessage());
            }
            return jsonResult;
        }
        /***
         *  telnet 操作
         * @param hostname
         * @param timeout in milliseconds
         * @return
         */
        public static JsonResult telnetResult(String hostname,Integer timeout){
            JsonResult jsonResult = new JsonResult();
            String[] split = hostname.split(":");
            try {
                Socket server = new Socket();
                InetSocketAddress address = new InetSocketAddress(split[0],Integer.parseInt(split[1]));
                server.connect(address, timeout);
                server.close();
                jsonResult.setMessage("telnet结果:success!");
            }catch (UnknownHostException e) {
                jsonResult.setCode(Constants.ResultCode.EXCEPTION);
                jsonResult.setMessage("telnet结果:UnknownHostException:"+e.getMessage());
            } catch (IOException e) {
                jsonResult.setCode(Constants.ResultCode.EXCEPTION);
                jsonResult.setMessage("telnet结果:IOException:"+e.getMessage());
            }
            return jsonResult;
        }
    
        /***
         *  telnet 操作
         * @param hostname
         * @param timeout in milliseconds
         * @return
         */
        public static JsonResult telnetResult(String hostname,Integer port,Integer timeout){
            JsonResult jsonResult = new JsonResult();
            try {
                Socket server = new Socket();
                InetSocketAddress address = new InetSocketAddress(hostname,port);
                server.connect(address, timeout);
                server.close();
                jsonResult.setMessage("telnet结果:success!");
            }catch (UnknownHostException e) {
                jsonResult.setCode(Constants.ResultCode.EXCEPTION);
                jsonResult.setMessage("telnet结果:UnknownHostException:"+e.getMessage());
            } catch (IOException e) {
                jsonResult.setCode(Constants.ResultCode.EXCEPTION);
                jsonResult.setMessage("telnet结果:IOException:"+e.getMessage());
            }
            return jsonResult;
        }
    
        public static void main(String[] args) {
            JsonResult jsonResult = telnetResult("192.168.1.80", 9092, 2000);
            System.out.println(jsonResult);
        }
    }

**日志输出到kafka**

import ch.qos.logback.classic.spi.ILoggingEvent;
    import ch.qos.logback.core.Appender;
    import ch.qos.logback.core.spi.AppenderAttachableImpl;
    import com.github.danielwegener.logback.kafka.KafkaAppenderConfig;
    import com.github.danielwegener.logback.kafka.delivery.FailedDeliveryCallback;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.KafkaException;
    import org.apache.kafka.common.serialization.ByteArraySerializer;
    
    import java.util.HashMap;
    import java.util.Iterator;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    /**
     * @ClassName KafkaAppenderAds
     * @Author dlh
     * @Date 2019/11/19 14:29
     * @Version 1.0
     **/
    
    public class KafkaAppenderAds<E> extends KafkaAppenderConfig<E> {
        /**
         * Kafka clients uses this prefix for its slf4j logging.
         * This appender defers appends of any Kafka logs since it could cause harmful infinite recursion/self feeding effects.
         */
        private static final String KAFKA_LOGGER_PREFIX = "org.apache.kafka.clients";
    
        public static volatile KafkaAppenderAds.LazyProducer lazyProducer = null;
        private final AppenderAttachableImpl<E> aai = new AppenderAttachableImpl<E>();
        private final ConcurrentLinkedQueue<E> queue = new ConcurrentLinkedQueue<E>();
        final static String LOCAL = "local-log";
    
        final static String KAFKA = "kafka-log";
    
        private static volatile Boolean telnetKafkaSucc = false;
        private final FailedDeliveryCallback<E> failedDeliveryCallback = new FailedDeliveryCallback<E>() {
            @Override
            public void onFailedDelivery(E evt, Throwable throwable) {
                aai.appendLoopOnAppenders(evt);
            }
        };
    
        public KafkaAppenderAds() {
            // setting these as config values sidesteps an unnecessary warning (minor bug in KafkaProducer)
            addProducerConfigValue(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
            addProducerConfigValue(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        }
    
        @Override
        public void doAppend(E e) {
            ensureDeferredAppends();
            if (e instanceof ILoggingEvent && ((ILoggingEvent)e).getLoggerName().startsWith(KAFKA_LOGGER_PREFIX)) {
                deferAppend(e);
            } else {
                super.doAppend(e);
            }
        }
    
        @Override
        public void start() {
            // only error free appenders should be activated
            if (!checkPrerequisites()) {
                return;
            }
    
            lazyProducer = new KafkaAppenderAds.LazyProducer();
    
            super.start();
        }
    
        @Override
        public void stop() {
            super.stop();
            if (lazyProducer != null && lazyProducer.isInitialized()) {
                try {
                    lazyProducer.get().close();
                } catch (KafkaException e) {
                    this.addWarn("Failed to shut down kafka producer: " + e.getMessage(), e);
                }
                lazyProducer = null;
            }
        }
    
        @Override
        public void addAppender(Appender<E> newAppender) {
            aai.addAppender(newAppender);
        }
    
        @Override
        public Iterator<Appender<E>> iteratorForAppenders() {
            return aai.iteratorForAppenders();
        }
    
        @Override
        public Appender<E> getAppender(String name) {
            return aai.getAppender(name);
        }
    
        @Override
        public boolean isAttached(Appender<E> appender) {
            return aai.isAttached(appender);
        }
    
        @Override
        public void detachAndStopAllAppenders() {
            aai.detachAndStopAllAppenders();
        }
    
        @Override
        public boolean detachAppender(Appender<E> appender) {
            return aai.detachAppender(appender);
        }
    
        @Override
        public boolean detachAppender(String name) {
            return aai.detachAppender(name);
        }
    
    
        @Override
        protected void append(E e) {
            if (telnetKafkaSucc == null){
                synchronized (this){
                    if (telnetKafkaSucc == null){
                        telnetKafkaSucc = false;
                    }
                }
            }
            if (!telnetKafkaSucc){
                return;
            }
            if (kafkaServerAddr == null){
                return;
            }
            final byte[] payload = encoder.doEncode(e);
            final byte[] key = keyingStrategy.createKey(e);
            final ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[],byte[]>(topic, key, payload);
            deliveryStrategy.send(lazyProducer.get(), record, e, failedDeliveryCallback);
        }
    
    
        public static void setTelnetKafkaSucc(boolean telnetKafkaSucc){
            KafkaAppenderAds.telnetKafkaSucc = telnetKafkaSucc;
        }
    
        public static Boolean getTelnetKafkaSucc(){
            return telnetKafkaSucc;
        }
    
        public static volatile String kafkaServerAddr;
        public static void setKafkaServerAddr(String kafkaServerAddr){
            KafkaAppenderAds.kafkaServerAddr = kafkaServerAddr;
        }
    
        protected Producer<byte[], byte[]> createProducer() {
            producerConfig.put("bootstrap.servers",KafkaAppenderAds.kafkaServerAddr);
            KafkaProducer<byte[], byte[]> kafkaProducer =
                    new KafkaProducer<>(new HashMap<String, Object>(producerConfig));
            return kafkaProducer;
        }
    
        private void deferAppend(E event) {
            queue.add(event);
        }
    
        // drains queue events to super
        private void ensureDeferredAppends() {
            E event;
    
            while ((event = queue.poll()) != null) {
                super.doAppend(event);
            }
        }
        public static KafkaAppenderAds.LazyProducer getLazyProducer(){
            return lazyProducer;
        }
        /**
         * Lazy initializer for producer, patterned after commons-lang.
         *
         * @see <a href="https://commons.apache.org/proper/commons-lang/javadocs/api-3.4/org/apache/commons/lang3/concurrent/LazyInitializer.html">LazyInitializer</a>
         */
        public class LazyProducer {
    
            private volatile Producer<byte[], byte[]> producer;
    
            public Producer<byte[], byte[]> get() {
                Producer<byte[], byte[]> result = this.producer;
                if (result == null) {
                    synchronized(this) {
                        result = this.producer;
                        if(result == null) {
                            this.producer = result = this.initialize();
                        }
                    }
                }
    
                return result;
            }
    
            protected Producer<byte[], byte[]> initialize() {
                Producer<byte[], byte[]> producer = null;
                try {
                    producer = createProducer();
                } catch (Exception e) {
                    addError("error creating producer", e);
                }
                return producer;
            }
    
            public boolean isInitialized() { return producer != null; }
    
            public boolean closeProducer(){
                if (isInitialized()){
                    this.producer.close();
                    this.producer = null;
                    return true;
                }
                return false;
            }
        }
    }

**定时调度**

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    
    /**
     * @ClassName ScheduledTelnet
     * @Author dlh
     * @Date 2019/11/14 16:07
     * @Version 1.0
     **/
    @Component
    public class ScheduledTelnetThread {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTelnetThread.class);
    
        private static final String DEFAULT_IP = "127.0.0.1:8000";
    
        @Value("${spring.kafka.bootstrap-servers:127.0.0.1:8000}")
        String kafkaServerAddr;
    
        Integer timeOut = 2000;
    
        private static final String SUCCESS_CODE = "200";
    
    
    
        @PostConstruct
        public void init(){
            TelnetThread telnetThread = new TelnetThread();
            telnetThread.setDaemon(true);
            telnetThread.start();
            KafkaTemplateInit kafkaTemplateInit = new KafkaTemplateInit();
            kafkaTemplateInit.setDaemon(true);
            kafkaTemplateInit.start();
        }
    
        class KafkaTemplateInit extends Thread{
            @Override
            public void run(){
                int count = 0;
                LOGGER.info("KafkaTemplateInit Thread start...");
                while (true){
                    count++;
                    if (count < 120 && DEFAULT_IP.equals(kafkaServerAddr)){
                        try {
                            sleep(500);
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                        continue;
                    }
                    if (!DEFAULT_IP.equals(kafkaServerAddr)){
                        KafkaAppenderAds.setKafkaServerAddr(kafkaServerAddr);
                        LOGGER.info("读取到的kafka地址:{}",kafkaServerAddr);
                        break;
                    }
                }
            }
        }
    
        class TelnetThread extends Thread{
            @Override
            public void run(){
                LOGGER.info("TelnetThread start..");
                while (true){
                    try {
                        if (kafkaServerAddr == null || DEFAULT_IP.equals(kafkaServerAddr)){
                            sleep(500);
                            continue;
                        }
                        telnet();
                        sleep(1000 * 60 * 2);
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }
    
            private void telnet(){
                try {
                    String[] addrSplit = kafkaServerAddr.split(":");
                    String hostName = addrSplit[0];
                    Integer port = Integer.parseInt(addrSplit[1]);
                    JsonResult jsonResult = PTUtil.telnetResult(hostName, port, timeOut);
                    LOGGER.info("telnet:{},result:{}",KafkaAppenderAds.getTelnetKafkaSucc(),jsonResult);
                    boolean telnetRes = SUCCESS_CODE.equals(jsonResult.getCode());
                    if (!telnetRes){
                        KafkaAppenderAds.getLazyProducer().closeProducer();
                        System.out.println("关闭kafka");
                    }
                    if (KafkaAppenderAds.getTelnetKafkaSucc() == telnetRes){
                        return;
                    }
                    KafkaAppenderAds.setTelnetKafkaSucc(telnetRes);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

**logback.xml**

<?xml version="1.0" encoding="UTF-8"?>
    <configuration>
        <springProperty scope="context" name="server-name" source="spring.application.name"
                        defaultValue="default-server"/>
        <contextName>poll-server</contextName>
        <property name="LOG_PATH" value="/home" />
        
        <property name="APP_DIR" value="app-server" />
        <property name="project_name" value="app-server"/>
        <property name="log.pattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS} ${LOG_LEVEL_PATTERN:-%5p} ${PID:-} [%thread]  [%-40.40logger{39}:%line] %m%n" />
    
        
        <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
            <encoder>
                <pattern>${log.pattern}</pattern>
            </encoder>
        </appender>
    
        
        <appender name="FILE_ERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
            
            <file>${LOG_PATH}/${APP_DIR}/${project_name}_log_error.log</file>
            
            <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
                <fileNamePattern>${LOG_PATH}/${APP_DIR}/error/${project_name}_log-error-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
                <maxFileSize>1GB</maxFileSize>
                <maxHistory>60</maxHistory>
                <totalSizeCap>20GB</totalSizeCap>
            </rollingPolicy>
            
            <append>true</append>
            
            <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
                <pattern>${log.pattern}</pattern>
                <charset>utf-8</charset>
            </encoder>
            
            <filter class="ch.qos.logback.classic.filter.LevelFilter">
                <level>ERROR</level>
                <onMatch>ACCEPT</onMatch>
                <onMismatch>DENY</onMismatch>
            </filter>
        </appender>
    
        
        <appender name="FILE_WARN" class="ch.qos.logback.core.rolling.RollingFileAppender">
            
            <file>${LOG_PATH}/${APP_DIR}/${project_name}_log_warn.log</file>
            
            <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
                <fileNamePattern>${LOG_PATH}/${APP_DIR}/warn/${project_name}_log-warn-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
                <maxFileSize>20MB</maxFileSize>
                <maxHistory>60</maxHistory>
                <totalSizeCap>20GB</totalSizeCap>
            </rollingPolicy>
            
            <append>true</append>
            
            <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
                <pattern>${log.pattern}</pattern>
                <charset>utf-8</charset>
            </encoder>
            
            <filter class="ch.qos.logback.classic.filter.LevelFilter">
                <level>WARN</level>
                <onMatch>ACCEPT</onMatch>
                <onMismatch>DENY</onMismatch>
            </filter>
        </appender>
    
        
        <appender name="FILE_INFO" class="ch.qos.logback.core.rolling.RollingFileAppender">
            
            <file>${LOG_PATH}/${APP_DIR}/${project_name}_log_info.log</file>
            
            <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
                <fileNamePattern>${LOG_PATH}/${APP_DIR}/info/${project_name}_log-info-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
                <maxFileSize>20MB</maxFileSize>
                <maxHistory>60</maxHistory>
                <totalSizeCap>20GB</totalSizeCap>
            </rollingPolicy>
            
            <append>true</append>
            
            <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
                <pattern>${log.pattern}</pattern>
                <charset>utf-8</charset>
            </encoder>
            
            <filter class="ch.qos.logback.classic.filter.LevelFilter">
                <level>INFO</level>
                <onMatch>ACCEPT</onMatch>
                <onMismatch>DENY</onMismatch>
            </filter>
        </appender>
    
        
        <appender name="FILE_DEBUG" class="ch.qos.logback.core.rolling.RollingFileAppender">
            
            <file>${LOG_PATH}/${APP_DIR}/${project_name}_log_debug.log</file>
            
            <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
                <fileNamePattern>${LOG_PATH}/${APP_DIR}/debug/${project_name}_log-debug-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
                <maxFileSize>20MB</maxFileSize>
                <maxHistory>60</maxHistory>
                <totalSizeCap>20GB</totalSizeCap>
            </rollingPolicy>
            
            <append>true</append>
            
            <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
                <pattern>${log.pattern}</pattern>
                <charset>utf-8</charset>
            </encoder>
            
            <filter class="ch.qos.logback.classic.filter.LevelFilter">
                <level>DEBUG</level>
                <onMatch>ACCEPT</onMatch>
                <onMismatch>DENY</onMismatch>
            </filter>
        </appender>
    
        <appender name="KafkaAppender"
                  class="com.shanhe.api.config.log.KafkaAppenderAds">
            <encoder
                    class="com.github.danielwegener.logback.kafka.encoding.LayoutKafkaMessageEncoder">
                <layout class="net.logstash.logback.layout.LogstashLayout">
                    
                    <includeContext>true</includeContext>
                    
                    <includeCallerData>true</includeCallerData>
                    
                    
                    
                    <fieldNames class="net.logstash.logback.fieldnames.ShortenedFieldNames" />
                </layout>
                <charset>UTF-8</charset>
            </encoder>
            
            <topic>
                app-logs
            </topic>
            <keyingStrategy
                    class="com.github.danielwegener.logback.kafka.keying.HostNameKeyingStrategy" />
            <deliveryStrategy
                    class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />
            <producerConfig>
                bootstrap.servers=192.168.1.80:9092
            </producerConfig>
            
            <appender-ref ref="CONSOLE" />
        </appender>
    
        
        <logger name="jdbc.sqltiming" level="Info"/>
    
        <logger name="org.mybatis" level="Info" />
        <logger name="org.apache.ibatis" level="Info" />
        <logger name="org.apache.ibatis.common.jdbc.ScriptRunner" level="Info" />
        <logger name="org.apache.ibatis.sqlmap.engine.impl.SqlMapClientDelegate" level="Info" />
        <logger name="java.sql.Connection" level="Info" />
        <logger name="java.sql.Statement" level="Info" />
        <logger name="java.sql.PreparedStatement" level="Info" />
        <logger name="java.sql.ResultSet" level="Info" />
    
        <logger name="org.springframework" level="Info" />
        <logger name="com.shanhe" level="Info" />
    
        <root level="Info">
            <appender-ref ref="FILE_ERROR" />
            <appender-ref ref="FILE_WARN" />
            <appender-ref ref="FILE_INFO" />
            <appender-ref ref="FILE_DEBUG" />
            <appender-ref ref="CONSOLE" />
            <appender-ref ref="KafkaAppender" />
        </root>
    </configuration>

**最后yml配置**

spring:
      kafka:
        bootstrap-servers:192.168.1.80:9092