收集了日志后,可以给别的地方做实时计算
很多应用程序使用Log4j记录日志,如何使用Kafka实时的收集与存储这些Log4j产生的日志呢?一种方案是使用其他组件(比如Flume,或者自己开发程序)实时监控这些日志文件,然后发送至Kafka。而另外一种比较便捷的方案是使用Kafka自带的Log4jAppender,在Log4j配置文件中进行相应的配置,即可完成将Log4j产生的日志实时发送至Kafka中。
本文以Kafka0.8.2为例,介绍KafkaLog4jAppender的配置方法:
log4j.properties文件内容如下:
1. log4j.rootLogger=INFO,console,KAFKA
2.
3. ## appender KAFKA
4. log4j.appender.KAFKA=kafka.producer.KafkaLog4jAppender
5. log4j.appender.KAFKA.topic=lxw1234
6. log4j.appender.KAFKA.brokerList=brokerNode1:9091,brokerNode2:9092
7. log4j.appender.KAFKA.compressionType=none
8. log4j.appender.KAFKA.syncSend=true
9. log4j.appender.KAFKA.layout=org.apache.log4j.PatternLayout
10. log4j.appender.KAFKA.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss}%-5p%c{1}:%L %%-%m%n
11.
12. ## appender console
13. log4j.appender.console=org.apache.log4j.ConsoleAppender
14. log4j.appender.console.target=System.err
15. log4j.appender.console.layout=org.apache.log4j.PatternLayout
16. log4j.appender.console.layout.ConversionPattern=%d (%t)[%p -%l]%m%n
注意:KAFKA appender的配置参数,和Kafka版本有关,具体参数可参考kafka.producer. KafkaLog4jAppender中的定义。
一个使用了Log4j记录日志的Java Application Demo:
1. package.lxw1234.kafka;
2.
3. import.apache.log4j.Logger;
4.
5. publicclassTestLog4j2Kafka{
6. privatestaticLogger=Logger.getLogger(TestLog4j2Kafka.class);
7.
8. publicstaticvoid(String[])throwsInterruptedException{
9. for(int=0;i <=10;++){
10. logger.info("This is Message ["++"] from log4j producer .. ");
11. Thread.sleep(1000);
12. }
13. }
14. }
先启动Kafka自带的consumer模拟脚本,消费Topic lxw1234的消息:
- cd $KAFKA_HOME/bin
- ./kafka-console-consumer.sh --zookeeper localhost:2181--topic lxw1234 --from-beginning
再运行上面的Java Demo程序,控制台打印的内容:
在Consumer控制台打印消费的消息:
如图所示,KafkaLog4jAppender已经将消息正常发送至Kafka。