13.kafka 收集日志的两种方式

小咪咪 2022-06-15 09:52 423阅读 0赞

方法一：

kafka官网已经提供了非常方便的log4j的集成包 kafka-log4j-appender，我们只需要简单配置log4j文件，就能收集应用程序log到kafka中。

#log4j.rootLogger=WARN,console,kafka
    log4j.rootLogger=INFO,console
    
    # for package com.demo.kafka, log would be sent to kafka appender.
    #log4j.logger.com.bigdata.xuele.streaming.SparkStreamingKmd*=info,kafka
    # appender kafka
    log4j.appender.kafka=kafka.producer.KafkaLog4jAppender
    log4j.appender.kafka.topic=${kafka.log.topic}
    # multiple brokers are separated by comma ",".
    log4j.appender.kafka.brokerList=${kafka.log.brokers}
    log4j.appender.kafka.compressionType=none
    log4j.appender.kafka.syncSend=false
    log4j.appender.kafka.layout=org.apache.log4j.PatternLayout
    #log4j.appender.kafka.layout.ConversionPattern=%d [%-5p] [%t] - [%l] %m%n
    log4j.appender.kafka.layout.ConversionPattern=[%d] %p %m (%c)%n
    
    # appender console
    log4j.appender.console=org.apache.log4j.ConsoleAppender
    log4j.appender.console.target=System.out
    log4j.appender.console.layout=org.apache.log4j.PatternLayout
    #log4j.appender.console.layout.ConversionPattern=%d [%-5p] [%t] - [%l] %m%n
    log4j.appender.console.layout.ConversionPattern=[%d] [%p] [%t] %m%n
    
    log4j.logger.org.eclipse.jetty=WARN
    log4j.logger.org.eclipse.jetty.util.component.AbstractLifeCycle=ERROR
    log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=INFO
    log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=INFO

注意，需要引入maven的依赖包：

<dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>0.8.2.1</version>
            </dependency>

非常简单，一个maven依赖加一个log4j配置文件即可，如果依然想写入log到本地 文件依然也是可以的，这种方式最简单快速，但是默认的的log日志是一行一行的纯文本，有些场景下我们可能需要json格式的数据。

方法二，java代码

package com.kafka;
    
    import java.util.Properties;
    import java.util.concurrent.Future;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    import org.apache.log4j.Level;
    import org.apache.log4j.Logger;
    
    import com.util.Config;
    
    public class KafkaUtil {
    	static {
    		// 修改kafka日志输出级别
    		Logger.getLogger("org.apache.kafka").setLevel(Level.OFF);
    	}
    
    	public static void sendMessage(String mes) {
    		KafkaProducer<String, String> producer = null;
    		try {
    			Properties properties = new Properties();
    			properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    			properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    			properties.put("bootstrap.servers", Config.getString("bootstrap.servers"));// 声明kafka
    			properties.put("acks", "all");// 声明kafka
    			properties.put("retries", 0);// 请求失败重试的次数
    			producer = new KafkaProducer<String, String>(properties);
    
    			ProducerRecord<String, String> record = new ProducerRecord<String, String>(
    					Config.getString("bootstrap.topic"), mes);
    			Future<RecordMetadata> future = producer.send(record);
    			System.out.println(future.get().offset() + "   " + future.get().partition());
    
    			// }
    		} catch (Exception e) {
    			e.printStackTrace();
    			// flag=false;
    			producer.close();
    		} finally {
    			producer.close();
    		}
    	}
    
    	public static void main(String[] args) {
    		while (true) {
    
    			KafkaUtil.sendMessage(
    					"null|127.0.0.1|国外|||||http://backserver/car/historyKeyAndHotKey.json|http://bd.2schome.net/|Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.19 Safari/537.36|http://backserver/car/historyKeyAndHotKey.json?keytype=0&|1470189803366185158");
    //			try {
    //				Thread.sleep(1000);
    //			} catch (InterruptedException e) {
    //
    //				// TODO Auto-generated catch block
    //				e.printStackTrace();
    //
    //			}
    		}
    
    	}
    
    }