最近完成一个简单的日志管理系统，拿出来跟大家分享一下！
主要实现的功能：
1、支持动态修改配置
2、实现统一的配置管理
3、支持文件输出、habse输出、mongodb输出
基于以上三点功能，我们下面详细说明
1、支持动态修改配置
说道支持这个功能，有个同事认为没有这个必要，他的观点是log4j的配置不需要经常变动，不需要支持这样的功能；本人的观点是“配置可以进行统一管理、而且正式机跟测试机的log4j的配置肯定会有一些差异的”，因此这个功能是必须的。
下面说一下实现
通过log4j提供的PropertyConfigurator类可以非常轻松的实现这个功能
代码如下
public class Log4jConfigListener {
private String log4jPath;
/**
* @param log4jPath the log4jPath to set
*/
public void setLog4jPath(String log4jPath) {
this.log4jPath = log4jPath;
}
/**
* 装载log4j配置文件
*
* @author mrh
* @DATE 2011-5-28
*/
public void load() {
// String path="config/log4j.properties";
System.out.println("log4j configfile path=" + log4j.toString());
PropertyConfigurator.configureAndWatch(log4j.toString(), 1000);// 间隔特定时间，检测文件是否修改，自动重新读取配置
}
}
Spring中的配置
<bean class="com.jl.net.log4j.config.Log4jConfigListener" init-method="load">
<property name="log4jPath" value="WEB-INF/classes/log4j/" /> <!--通过zookeeper实现统一配置的dataId-->
</bean>
2、统一配置管理
统一配置管理，是通过zookeeper实现的，配置形式与log4j.properties配置一致；
配置如下
log4j.rootLogger=INFO,console,HbaseAppender
#console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{ABSOLUTE} %5p %c{1}\:%L - %m%n
#hbase
log4j.appender.HbaseAppender=com.jl.net.log4j.hbase.HbaseAppender
log4j.appender.HbaseAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.HbaseAppender.sysName=tuid
log4j.appender.HbaseAppender.serverIp=Y
log4j.appender.HbaseAppender.zookeeper_ip=10.1.18.100,10.1.18.103,10.1.18.102
log4j.appender.HbaseAppender.zookeeper_port=2181
log4j.appender.HbaseAppender.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss sss} %c.%t:%L %-5p %x %m%n
Spring中的配置
<bean class="com.jl.net.log4j.config.Log4jConfigListener" init-method="load">
<property name="log4jPath" value="cfg/jlcloud.uid.log4j" /> <!--通过zookeeper实现统一配置的dataId-->
</bean>
3、支持文件、habse、mongodb的输出
log4j本身就支持文件输出，mongodb的输出需要借助log4mongo-Java这个项目，maven的坐标是
<dependency>
<groupId>org.log4mongo</groupId>
<artifactId>log4mongo-java</artifactId>
<version>0.7.4</version>
<optional>true</optional>
</dependency>
下面介绍一下如何实现hbase的输出，通过继承log4j的AppenderSkeleton类，同时实现Runnable接口，来实现：
代码如下
package com.yck.worm.hbase;
import java.io.IOException;
import java.net.InetAddress;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.AppenderSkeleton;
import org.apache.log4j.spi.LoggingEvent;
import com.yck.worm.logging.WLogger;
/**
* 将日志 输出到HBase里
* @author mrh
*
*/
public class HbaseAppender extends AppenderSkeleton implements Runnable {
private static final WLogger LOGGER = WLogger.getLogger(HbaseAppender.class);
private int batchSize = -1;
private int period = 1000;
private String hbLogName = "jl_logs";
private String hbLogFamily = "logs";
private Queue<LoggingEvent> loggingEvents;
private ScheduledExecutorService executor;
private ScheduledFuture<?> task;
private Configuration conf;
private HTableInterface htable;
private HBaseAdmin admin;
private HConnection connection;
private String zookeeper;
private String sysName;
/**
* if serverIp = Y then out put server's IP to hbase
*/
private String serverIp;
/**
* log4j初始设置，启动日志处理计划任务
*/
@Override
public void activateOptions() {
try {
super.activateOptions();
// 创建一个计划任务，并自定义线程名
executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("HbaseAppender"));
// 日志队列
loggingEvents = new ConcurrentLinkedQueue<LoggingEvent>();
// 启动计划任务，如果run函数有异常任务将中断！
task = executor.scheduleWithFixedDelay(this, period, period,TimeUnit.MILLISECONDS);
if ("Y".equalsIgnoreCase(this.serverIp)) {
InetAddress addr = InetAddress.getLocalHost();
this.serverIp = addr.getHostAddress().toString();//获得本机IP
}
LOGGER.info("ActivateOptions ok!");
} catch (Exception e) {
LOGGER.error("Error during activateOptions: " , e);
}
}
/**
* 初始HBASE
*
* @return
*/
private boolean initHbase() {
try {
if (conf == null) {
LOGGER.info("initHbase ... zookeeper address is " + this.zookeeper);
if (this.zookeeper == null || "".equals(this.zookeeper)) {
throw new IllegalArgumentException("the zookeeper address for jlog4j's hadoop initial can not be null!");
}
String ip = this.zookeeper.split(":")[0];
String port = this.zookeeper.split(":")[1];
conf = HBaseConfiguration.create();
conf.set(HConstants.ZOOKEEPER_QUORUM, ip);
conf.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, Integer.parseInt(port));
this.admin = new HBaseAdmin(conf);
if (!this.admin.tableExists(Bytes.toBytes(hbLogName))) {
HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(hbLogName));
HColumnDescriptor columnDescriptor = new HColumnDescriptor(hbLogFamily.getBytes());
tableDescriptor.addFamily(columnDescriptor);
this.admin.createTable(tableDescriptor);
}
this.connection = HConnectionManager.createConnection(conf);
this.htable = this.connection.getTable(hbLogName.getBytes());
}
LOGGER.error("Init Hbase success !");
return true;
} catch (Exception e) {
task.cancel(false);
executor.shutdown();
LOGGER.error("Init Hbase fail !", e);
return false;
}
}
@Override
public void run() {
if (conf == null || htable == null) {
initHbase();
}
try {
if (this.batchSize < 0 && loggingEvents.size() > 0 ) {
this.output();
} else
// 日志数据超出批量处理大小
if (loggingEvents.size() > 0 && loggingEvents.size() > this.batchSize) {
this.output();
}
} catch (Exception e) {
LOGGER.error("HbaseAppender Error run ", e);
}
}
/**
* 输出日志
* @throws IOException
*/
private void output() throws IOException {
LoggingEvent event;
List<Put> logs = new ArrayList<Put>();
// 循环处理日志队列
while ((event = loggingEvents.poll()) != null) {
try {
// 写日志内容
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SSS");
Date date = new Date(event.getTimeStamp());
// 创建日志并指定ROW KEY
StringBuffer key = new StringBuffer("");
if (this.sysName != null)
key.append(this.sysName).append("#");
key.append(event.getLoggerName()).append("#");
key.append(event.getLevel()).append("#");
key.append(sdf.format(date)).append("#");
key.append(UUID.randomUUID().toString().replace("-", ""));
Put log = new Put(key.toString().getBytes());
if (this.sysName != null)
log.add(hbLogFamily.getBytes(), "sysName".getBytes(), this.sysName.getBytes());
if (this.serverIp != null)
log.add(hbLogFamily.getBytes(), "serverIp".getBytes(), this.serverIp.getBytes());
log.add(hbLogFamily.getBytes(), "LoggerName".getBytes(), event.getLoggerName().getBytes());
log.add(hbLogFamily.getBytes(), "level".getBytes(), event.getLevel().toString().getBytes());
log.add(hbLogFamily.getBytes(), "datetime".getBytes(), sdf.format(date).getBytes());
log.add(hbLogFamily.getBytes(), "message".getBytes(), event.getMessage().toString().getBytes());
logs.add(log);
} catch (Exception e) {
LOGGER.error("Error logging put ", e);
}
}
// 批量写入HBASE
if (logs.size() > 0)
htable.put(logs);
}
/**
* 日志事件
*
* @param loggingEvent
*/
@Override
protected void append(LoggingEvent loggingEvent) {
try {
populateEvent(loggingEvent);
// 添加到日志队列
loggingEvents.add(loggingEvent);
} catch (Exception e) {
LOGGER.error("Error populating event and adding to queue", e);
}
}
/**
* 事件测试
*
* @param event
*/
protected void populateEvent(LoggingEvent event) {
event.getThreadName();
event.getRenderedMessage();
event.getNDC();
event.getMDCCopy();
event.getThrowableStrRep();
event.getLocationInformation();
}
@Override
public void close() {
try {
if (this.task != null)
this.task.cancel(false);
if (this.executor != null)
this.executor.shutdown();
if (this.connection != null)
this.connection.close();
if (this.admin != null)
this.admin.close();
if (this.htable != null)
this.htable.close();
} catch (IOException e) {
LOGGER.error("Error close ", e);
}
}
@Override
public boolean requiresLayout() {
return true;
}
// 设置每一批日志处理数量
public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
}
/**
* 设置计划任务执行间隔
*
* @param period
*/
public void setPeriod(int period) {
this.period = period;
}
/**
* 设置日志存储HBASE表名
*
* @param hbLogName
*/
public void setHbLogName(String hbLogName) {
this.hbLogName = hbLogName;
}
/**
* 日志表的列族名字
*
* @param hbLogFamily
*/
public void setHbLogFamily(String hbLogFamily) {
this.hbLogFamily = hbLogFamily;
}
/**
* @param zookeeper the zookeeper to set
*/
public void setZookeeper(String zookeeper) {
this.zookeeper = zookeeper;
}
/**
* @param sysName the sysName to set
*/
public void setSysName(String sysName) {
this.sysName = sysName;
}
/**
* @param serverIp the serverIp to set
*/
public void setServerIp(String serverIp) {
this.serverIp = serverIp;
}
}
收工，有些粗糙，将就吧！
附带源码，见附件
相关推荐
dlog分布式日志系统
一个Java开发的轻量级分布式日志标记追踪神器 一个Java开发的轻量级分布式日志标记追踪神器 一个Java开发的轻量级分布式日志标记追踪神器 一个Java开发的轻量级分布式日志标记追踪神器 一个Java开发的轻量级...
分布式日志处理系统调研报告，主要针对ELk的使用现状进行分析，对技术的成熟度和现有使用厂家的使用情况进行总结。
分布式日志管理分布式日志管理
基于FlumeKafkaSpark-的分布式日志流处理系统的设计与实现
基于Python内建库的分布式日志系统设计与实现.pdf
DooS：分布式日志采集存储微服务，宋鹏威，徐鹏，随着部署在云上应用不断增加，无论是系统运行期间，还是应用使用期间都会产生大量日志，此外还有些应用有海量数据存储需求。为了
基于BeatWatcher的分布式日志采集系统的设计与实现，陈沅星，鄂海红，日志数据的采集，在排查和分析系统运行问题以及运行状况等方面起着极其重要的作用，也是其最基础的一步。但是在对日志数据进行采
本例子是用ServiceStack.Redis+log4net做的一个分布式记日志的功能，使用场景：分布式系统中的各个服务向redis服务添加日志，服务定时轮询生成记录日志到数据库或文本中。如果你的系统所有的方法调记录都需要记录的...
第三章日志索引分布式存储策略.....................................................................……27 3.1数据库索引概述..................................................................................
分布式日志分析系统的设计与实现
深入介绍Flume众多更加有用的组件的细节信息，包括用于即时数据记录持久化的重要的文件通道、用于缓存并将数据写到HDFS中的HDFS接收器，以及Hadoop分布式文件系统。对于Flume各个架构组件（源、通道、接收器、通道...
#资源达人分享计划#
一个java分布式日志组件，支持百亿级别，日志从搜集到查询，不用去文件中翻阅日志方便快捷，支持查询一个调用链的日志，支持链路追踪，查看调用链耗时情况，在分布式系统中也可以查询关联日志，能够帮助快速定位问题...
一、ELK应用背景： 二、ELK平台概述： 三、ELK核心组件： 四、案例：部署ELK分布式日志分析平台： 五、案例：应用ELK日志分析平台进行分析nginx日志：
基于thrift+scribe实现分布式日志收集，并能够基于log4j进行集成
TANK 是一个非常高性能的分布式日志，部分受到 Kafka 以及其他类似的服务和技术的启发。特性：高性能严格的消息编码简单的设计基于消息键和将来基于可编程逻辑的压缩强大的 CLI 工具，用于管理、查询、设置消、跨...
日志管理系统 面向分布式日志管理系统开发的日志管理系统
基于Hadoop集群的分布式日志分析系统研究