应用程序的日志信息可通过log4j同步到Kafka中，即：

客户端——>后台服务应用程序——>Kafka集群

代码如下：

1.pom.xml文件导入log4j与kafka的集成依赖库

       

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-log4j-appender</artifactId>
            <version>1.0.1</version>
        </dependency>


2.新建一个log4j.properties文件，并添加appender配置到kafka中

log4j.rootLogger=info,stdout,kafka0
 
# stdout配置
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%-d{yyyy-MM-dd HH\:mm\:ss} [%p]-[%c] %m%n
 
 
# kafka0配置
#定义一个名为kafka 为Appender
log4j.appender.kafka0=org.apache.kafka.log4jappender.KafkaLog4jAppender
#指定日志写入到Kafka的主题
log4j.appender.kafka0.topic=kafka-log4j
#制定连接kafka的地址
log4j.appender.kafka0.brokerList=192.168.23.128:9092
#压缩方式，默认为none
log4j.appender.kafka0.compressionType=none
#指定Producer发送消息的方式，默认是false，即异步发送
log4j.appender.kafka0.syncSend=true
#指定日志级别
log4j.appender.kafka0.Threshold=info
log4j.appender.kafka0.layout=org.apache.log4j.PatternLayout
log4j.appender.kafka0.layout.ConversionPattern=%-d{yyyy-MM-dd HH\:mm\:ss} [%p]-[%c] %m%n


3.编写应用程序Log4jProducer

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
 
public class Log4jProducer {
    private static final Logger LOG = Logger.getLogger(Log4jProducer.class);
 
    public static void main(String[] args) {
        LOG.setLevel(Level.INFO);
        LOG.debug("这是一条debug级别的日志！");
        LOG.info("这是一条info级别的日志！");
        LOG.error("这是一条error级别的日志！");
        LOG.fatal("这是一条fatal级别的日志！");
    }
}


4.创建kafka topic用来存储程序日志信息

bin/kafka-topics.sh --zookeeper 192.168.23.128:2181 --create --topic kafka-log4j --partitions 1 --replication-factor 1


并启动一个客户端消费者：

 

bin/kafka-console-consumer.sh  --bootstrap-server 192.168.23.128:9092 --topic kafka-log4j --from-beginning


5.运行应用程序Log4jProducer，并看kafka consumer对应的输出

 

 
2018-08-25 17:50:11 [INFO]-[Log4jProducer] 这是一条info级别的日志！
 
2018-08-25 17:50:11 [ERROR]-[Log4jProducer] 这是一条error级别的日志！
 
2018-08-25 17:50:11 [FATAL]-[Log4jProducer] 这是一条fatal级别的日志！


具体输出什么级别的信息到Kafka集群中，可通过log4j.appender.kafka0.Threshold以及应用程序中的LOG.setLevel()配合设置。

应用程序console端的输出如下，比Kafka端多了一个ProducerConfig的配置信息，这是在log4j中启动Kafka Producer生成的，所以kafka topic中看不到该信息。

"C:\Program Files\Java\jdk1.8.0_171\bin\java.exe" -javaagent:C:\sf\ideaIU-2018.1.4.win\lib\idea_rt.jar=57230:C:\sf\ideaIU-2018.1.4.win\bin -Dfile.encoding=UTF-8 -classpath "C:\Program Files\Java\jdk1.8.0_171\jre\lib\charsets.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\deploy.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\access-bridge-64.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\cldrdata.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\dnsns.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\jaccess.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\jfxrt.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\localedata.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\nashorn.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\sunec.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\sunjce_provider.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\sunmscapi.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\sunpkcs11.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\ext\zipfs.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\javaws.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\jce.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\jfr.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\jfxswt.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\jsse.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\management-agent.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\plugin.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\resources.jar;C:\Program Files\Java\jdk1.8.0_171\jre\lib\rt.jar;F:\kafka\workspace\kafkaLog4jDemo\target\classes;C:\Users\sang\.m2\repository\org\apache\kafka\kafka-log4j-appender\1.0.1\kafka-log4j-appender-1.0.1.jar;C:\Users\sang\.m2\repository\org\apache\kafka\kafka-clients\1.0.1\kafka-clients-1.0.1.jar;C:\Users\sang\.m2\repository\org\lz4\lz4-java\1.4\lz4-java-1.4.jar;C:\Users\sang\.m2\repository\org\xerial\snappy\snappy-java\1.1.4\snappy-java-1.1.4.jar;C:\Users\sang\.m2\repository\org\slf4j\slf4j-api\1.7.25\slf4j-api-1.7.25.jar;C:\Users\sang\.m2\repository\org\slf4j\slf4j-log4j12\1.7.25\slf4j-log4j12-1.7.25.jar;C:\Users\sang\.m2\repository\log4j\log4j\1.2.17\log4j-1.2.17.jar" Log4jProducer
2018-08-25 17:57:18 [INFO]-[org.apache.kafka.clients.producer.ProducerConfig] ProducerConfig values: 
	acks = 1
	batch.size = 16384
	bootstrap.servers = [192.168.23.128:9092]
	buffer.memory = 33554432
	client.id = 
	compression.type = none
	connections.max.idle.ms = 540000
	enable.idempotence = false
	interceptor.classes = null
	key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
	linger.ms = 0
	max.block.ms = 60000
	max.in.flight.requests.per.connection = 5
	max.request.size = 1048576
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
	receive.buffer.bytes = 32768
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 0
	retry.backoff.ms = 100
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	send.buffer.bytes = 131072
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
	ssl.endpoint.identification.algorithm = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLS
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	transaction.timeout.ms = 60000
	transactional.id = null
	value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
 
2018-08-25 17:57:19 [INFO]-[org.apache.kafka.common.utils.AppInfoParser] Kafka version : 1.0.1
2018-08-25 17:57:19 [INFO]-[org.apache.kafka.common.utils.AppInfoParser] Kafka commitId : c0518aa65f25317e
2018-08-25 17:57:19 [INFO]-[Log4jProducer] 这是一条info级别的日志！
2018-08-25 17:57:19 [ERROR]-[Log4jProducer] 这是一条error级别的日志！
2018-08-25 17:57:19 [FATAL]-[Log4jProducer] 这是一条fatal级别的日志！


到此，一个简单的应用程序日志信息保存到Kafka中的应用就写好了。代码结构如下：

 

 

其它：log4j的info和error分开存储
应用程序同步到远程机器会增加额外的带宽占用，且远程机器故障会影响到本机，我们可把重要信息如error保存在Kafka中，其它保存在本地文件中(这里只是简单实践下这个功能，实际工作中很少这样做)。这样做只需修改log4j.properties文件即可。

log4j.properties

#Log4j info和error输出到不同目的地
 
log4j.rootLogger=info,infolog,kafka0
 
log4j.appender.infolog=LogAppender  #自定义的Appender，让每个级别类型的日志只提取与之优先级（ Priority ）相等的日志
log4j.appender.infolog.layout=org.apache.log4j.PatternLayout
log4j.appender.infolog.layout.ConversionPattern=%-d{yyyy-MM-dd HH\:mm\:ss} [%p]-[%c] %m%n
log4j.appender.infolog.Threshold=INFO
log4j.appender.infolog.append=true
log4j.appender.infolog.File=F:\\kafka\\workspace\\kafkaLog4jDemo\\logs\\kafka-log4j-info.log
 
 
#定义一个名为kafka 为Appender
log4j.appender.kafka0=org.apache.kafka.log4jappender.KafkaLog4jAppender
#指定日志写入到Kafka的主题
log4j.appender.kafka0.topic=kafka-log4j
#制定连接kafka的地址
log4j.appender.kafka0.brokerList=192.168.23.128:9092
#压缩方式，默认为none
log4j.appender.kafka0.compressionType=none
#指定Producer发送消息的方式，默认是false，即异步发送
log4j.appender.kafka0.syncSend=true
#指定日志级别
log4j.appender.kafka0.Threshold=ERROR
log4j.appender.kafka0.layout=org.apache.log4j.PatternLayout
log4j.appender.kafka0.layout.ConversionPattern=%-d{yyyy-MM-dd HH\:mm\:ss} [%p]-[%c] %m%n


自定义的LogAppender代码如下，继承DailyRollingF ileAppender 类， 重写isAsSevereAsThreshol()方法，只保存当前级别的日志信息，如info则只保存info级别，而不是小于等于info级别的信息。

import org.apache.log4j.DailyRollingFileAppender;
import org.apache.log4j.Priority;
 
public class LogAppender extends DailyRollingFileAppender {
    @Override
    public  boolean isAsSevereAsThreshold(Priority priority){
        return this.getThreshold().equals(priority);
    }
}

 

参考文档：

1.书《kafka入门与实践》

2.log4j教程 https://www.yiibai.com/log4j/
--------------------- 
作者：wateryouyouyou 
来源：CSDN 
原文：https://blog.csdn.net/wateryouyo/article/details/82053371 
版权声明：本文为博主原创文章，转载请附上博文链接！

Logo
Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南