我对Apache·Kafka和log4j很陌生。我正试图把我的日志信息发送到Kafka。这是我的log4j属性文件
log4j.rootLogger=INFO, 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} %-5p %c{1}:%L %% %m%n
log4j.appender.KAFKA=kafka.producer.KafkaLog4jAppender
log4j.appender.KAFKA.BrokerList=localhost:9092
log4j.appender.KAFKA.Topic=kfkLogs
log4j.appender.KAFKA.SerializerClass=kafka.producer.DefaultStringEncoder
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
log4j.logger.logGen=DEBUG, KAFKA
但是,我无法在我的消费者中接收任何消息。我已经测试了一些其他生产者代码的消费者,它的作品很好。
还有,我收到这个警告
log4j:WARN No such property [serializerClass] in kafka.producer.KafkaLog4jAppender.
编辑
下面是生成日志消息的代码
package logGen;
import org.apache.log4j.Logger;
public class TestLog4j {
static Logger log = Logger.getLogger(TestLog4j.class.getName());
public static void main(String[] args) {
log.debug("Debug message");
log.info("Info message");
log.error("Error Message");
log.fatal("Fatal Message");
log.warn("Warn Message");
log.trace("Trace Message");
}
}
另外,如果我使用
log4j.appender.KAFKA=org.apache.log4j.DailyRollingFileAppender
log4j.appender.KAFKA.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.KAFKA.File=logs/server.log
我可以在server.log文件中看到日志消息
谢谢大家的建议。我想我看到的怪异行为可能和我的Kafka设计有关。下面是我用来启动kafka服务器的server.properties文件的内容。你能看出有什么奇怪的地方吗?
broker.id=0
port=9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/Users/xyz/kafka/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
delete.topic.enable=true
4条答案
按热度按时间tzxcd3kk1#
你需要加上
KAFKA
给你的log4j.rootLogger
这样地:这将添加
KAFKA
附加到根记录器。d8tt03nd2#
我已经查看了kafkalog4jappender.scala的源代码,下面是kafka log4j appender在0.8.2.1版本之后的有效和详尽的属性:topic、brokerlist、compressiontype、requirednumacks、syncsend。
为我工作的log4j.properties如下:
k4emjkb13#
log4j.appender.kafka.serializerclass=kafka.producer.defaultstringencoder
你能试试吗
log4j.appender.kafka.serializer=kafka.producer.defaultstringencoder
相反呢?
我相信发送msg的异步模式是通过批处理来实现的,因此会有延迟,您尝试过使用sync发送吗?
woobm2wo4#
我必须详细说明
log4j.appender.kafka.producer.type=异步
log4j.logger.loggen.testlog4j=跟踪,Kafka
这似乎奏效了。但是我经历了10-30秒的延迟。具体来说,如果我现在发布并且可以在消费者中看到消息,那么下一次发布必须在30秒之后,否则我在消费者中看不到任何东西。你知道为什么会这样吗?也许这是个问题?