Demo及相关软件下载：https://download.csdn.net/download/wqc19920906/10342360

很多应用程序使用Log4j记录日志，如何使用Kafka实时的收集与存储这些Log4j产生的日志呢？一种方案是使用其他组件（比如Flume，或者自己开发程序）实时监控这些日志文件，然后发送至Kafka。而另外一种比较便捷的方案是使用Kafka自带的Log4jAppender，在Log4j配置文件中进行相应的配置，即可完成将Log4j产生的日志实时发送至Kafka中。

本文以Kafka0.8.2为例，介绍KafkaLog4jAppender的配置方法：

log4j.properties文件内容如下：

 ### 设置###
log4j.rootLogger=INFO,console,KAFKA
 
## appender KAFKA
log4j.appender.KAFKA=kafka.producer.KafkaLog4jAppender
log4j.appender.KAFKA.topic=test
log4j.appender.KAFKA.brokerList=127.0.0.1:9092
log4j.appender.KAFKA.compressionType=none
log4j.appender.KAFKA.syncSend=true
log4j.appender.KAFKA.layout=org.apache.log4j.PatternLayout
log4j.appender.KAFKA.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% - %m%n
 
## appender console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d (%t) [%p - %l] %m%n

注意：KAFKA appender的配置参数，和Kafka版本有关，具体参数可参考kafka.producer. KafkaLog4jAppender中的定义。

一个使用了Log4j记录日志的Java Application Demo:

public class App {

	private static Logger logger = Logger.getLogger(App.class);

	public static void main(String[] args) {

		for (int i = 0; i <= 1000; i++) {
			logger.info("This is Message [" + i + "] from log4j producer .. ");
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}

首先要启动zookeeper, 再次启动Kafka, 然后创建消费者：

kafka-console-consumer.bat --zookeeper localhost:2181 --topic test --from-beginning

再运行上面的Java Demo程序，控制台打印的内容：

在Consumer控制台打印消费的消息：

如图所示，KafkaLog4jAppender已经将消息正常发送至Kafka。