NoSql存储日志数据之Spring+Logback+Hbase深度集成
关键词：nosql, spring logback, logback hbase appender
技术框架：spring-data-hadoop, logback
一些比较重要的日志信息需要经常查看，比如用户行为日志，报错或关键业务日志数据然而同一系统多结点运行时这个工作会变的非常繁琐。
本例借用Logback日志框架和Hbase数据库来解决这一问题。
主要功能：
- 所有结点日志数据可通过配置同步到一个Hbase数据库
- 与Spring整合，全局共享一个Hbase操作实例，动态为某日志添加Appender
- 存储的日志数据可指定日志和日志级别，日志过滤
- Key-Value方式存储，可指定Value的生成格式
Hbase的操作采用的是Spring-Hadoop中Hbase部分实现，没有直接引用Spring-Hadoop而是只提取Hbase实现部分，原因是Hbase只是其中很小一部分又不想自己封装，因此直接提取使用，封装功能包括Hbase配置和HbaseTemplate等，相关源码请参考Spring-Data-Hadoop下Hbase部分！
关于Logback没有直接使用Logback的配置文件来配置HbaseAppender的原因是想与Spring整合使用Spring的Hbase-bean实例，不用单为Logback建一个Hbase实现等等。
使用相关配置
本文采用的NoSql数据库为Hbase，并且只有Hbase的Logback Appender实现类，可参考实现其它类型的Appender。
Logback Nosql Appender工厂类
在这里指定用哪种类型的NoSql数据库操作对像和对应的LOGBACK APPENDER操作类。
<beanid="logbackNosqlFactory"class="b2gonline.wap.logback.NosqlAppenderFactoryBean"><!--数据库操作类--><propertyname="template"ref="hbaseTemplate"/><!--日志存储表名--><propertyname="tbname"value="waplogdata"/><!--Logback Appender 类全路径--><propertyname="appenderPaht"value="b2gonline.wap.logback.hbase.HbaseAppender"/></bean>Logback Appender配置
在这里配置LOGGER和APPENDER的对应关系和详细的APPENDER配置，引用的APPENDER从上面配置的Logback Nosql Appender工厂类获得。
<!--单个LOGGER配置--><beanid="hbaseappender1"class="b2gonline.wap.logback.SpringLogbackBean"lazy-init="false"><propertyname="appenderName"value="hbaseappender1"/><propertyname="logLevel"value="INFO"/><propertyname="filterLevel"value="INFO"/><propertyname="logName"value="logbackhbaseappendertest"/><propertyname="appender"ref="logbackNosqlFactory"/></bean><!--多个LOGGER配置--><beanid="hbaseappender2"class="b2gonline.wap.logback.SpringLogbackBean"lazy-init="false"><propertyname="appenderName"value="hbaseappender2"/><propertyname="filterLevel"value="WARN"/><propertyname="logName"><map><entrykey="logbackhbaseappendertest2"value="WARN"/><entrykey="logbackhbaseappendertest3"value="WARN"/></map></property><propertyname="appender"ref="logbackNosqlFactory"/></bean>JAVA代码实现
NosqlAppenderFactoryBean
不同NOSQL数据库实现的LOGBACK APPENDER工厂类
import b2gonline.wap.logback.hbase.HbaseAppender;import org.springframework.beans.factory.FactoryBean;import org.springframework.util.Assert;/**
* NosqlAppenderFactoryBean工厂类，根据Nosql类型返回实例
*/publicclassNosqlAppenderFactoryBeanimplementsFactoryBean<NoSqlAppender>{privateObject template;privateString tbname;privateString appenderPaht;/**
* Appender类路径，实例化不同类型Appender实例
* @param appender
*/publicvoid setAppenderPaht(String appender){this.appenderPaht = appender;}/**
* 指定类型数据库操作类
* @param template
*/publicvoid setTemplate(Object template){this.template = template;}/**
* 数据存储表名
* @param tbname
*/publicvoid setTbname(String tbname){this.tbname = tbname;}/**
* 根据数据库类型返回Appender实例
*
* @return
* @throws Exception
*/@OverridepublicNoSqlAppender getObject()throwsException{//校验配置Assert.notNull(template);Assert.notNull(appenderPaht);Assert.notNull(tbname);//生成实例Class<?> appenderClass =Class.forName(appenderPaht);StringSCname= appenderClass.getSuperclass().getSimpleName();if(SCname.equals("NoSqlAppender")){NoSqlAppender hbaseAppender =(NoSqlAppender) appenderClass.newInstance();
hbaseAppender.setTemplate(template);
hbaseAppender.setTbname(tbname);return hbaseAppender;}else{thrownewIllegalArgumentException(appenderPaht +"is not NoSqlAppender subclass!");}}@OverridepublicClass<?> getObjectType(){returnHbaseAppender.class;}@Overridepublicboolean isSingleton(){returnfalse;}}SpringLogbackBean
Logback Appender与Spring整合类，参考上面第二部分配置
import ch.qos.logback.classic.Level;import ch.qos.logback.classic.Logger;import ch.qos.logback.classic.LoggerContext;import ch.qos.logback.classic.filter.LevelFilter;import ch.qos.logback.core.spi.FilterReply;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.InitializingBean;import org.springframework.util.Assert;import java.util.Map;/**
* Logback Appender与Spring整合类
*/publicclassSpringLogbackBeanimplementsInitializingBean{
org.slf4j.Logger _logger =LoggerFactory.getLogger(this.getClass());privateLevel logLevel =Level.INFO;privateString appenderName ="NoSqlAppender";privateNoSqlAppender appender;privateObject logName ="root";privateLevel filterLevel =Level.INFO;privateboolean useFilterLevel =true;privateboolean additiveAppender =true;privateString pattern ="%d{HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n";/**
* Appender 名称
*
* @param appenderName
*/publicvoid setAppenderName(String appenderName){this.appenderName = appenderName;}/**
* 设置存储的日志级别，默认是INFO
*
* @param logLevel
*/publicvoid setLogLevel(Level logLevel){this.logLevel = logLevel;}/**
* 设置Logback appender
*
* @param appender
*/publicvoid setAppender(NoSqlAppender appender){this.appender = appender;}/**
* 设置需要记录的日志名，默认是root
*
* @param logName
*/publicvoid setLogName(Object logName){this.logName = logName;}/**
* 使用过滤器过滤的日志级别，默认INFO
*
* @param filterLevel
*/publicvoid setFilterLevel(Level filterLevel){this.filterLevel = filterLevel;}/**
* 是否累加Appender(继承root appender)，默认 true
*
* @param additiveAppender
*/publicvoid setAdditiveAppender(boolean additiveAppender){this.additiveAppender = additiveAppender;}/**
* 是否使用日志过滤，其它级别日志数据交给继承来的APPENDER处理
*
* @param useFilterLevel
*/publicvoid setUseFilterLevel(boolean useFilterLevel){this.useFilterLevel = useFilterLevel;}/**
* 设置日志格式
*
* @param pattern
*/publicvoid setPattern(String pattern){this.pattern = pattern;}@Overridepublicvoid afterPropertiesSet()throwsException{Assert.notNull(appender,"property `appender` must set!");
buildAppender();//多LOGGER支持配置if(logName.getClass().getSimpleName().equals("LinkedHashMap")){Map<String,String> loggers =(Map) logName;for(Map.Entry<String,String> log : loggers.entrySet()){Logger logger =(Logger)LoggerFactory.getLogger(log.getKey());
logger.setLevel(Level.toLevel(log.getValue()));
logger.setAdditive(additiveAppender);
logger.addAppender(appender);
_logger.debug("Set appender: {} to logger: {} ", appender.getName(), log.getKey());}}//单个配置else{Logger logger =(Logger)LoggerFactory.getLogger(logName.toString());//将appender添加到指定logger
logger.setLevel(logLevel);
logger.setAdditive(additiveAppender);
logger.addAppender(appender);
_logger.debug("Set appender: {} to logger: {} ", appender.getName(), logName);}}privatevoid buildAppender(){LoggerContext loggerContext =(LoggerContext)LoggerFactory.getILoggerFactory();//启用级别过滤，适用场景：把级别为 warn 是放入数据库。if(useFilterLevel){LevelFilter levelFilter =newLevelFilter();
levelFilter.setContext(loggerContext);
levelFilter.setLevel(filterLevel);
levelFilter.setOnMatch(FilterReply.ACCEPT);
levelFilter.setOnMismatch(FilterReply.DENY);
levelFilter.start();
appender.addFilter(levelFilter);}//设置appender相关属性
appender.setName(appenderName);
appender.setPattern(pattern);
appender.setContext(loggerContext);
appender.start();}}NoSqlAppender
NoSql Appender 基础类，不同NOSQL数据库依赖此类实现不同APPENDER。
import ch.qos.logback.classic.PatternLayout;import ch.qos.logback.classic.spi.ILoggingEvent;import ch.qos.logback.core.UnsynchronizedAppenderBase;import ch.qos.logback.core.spi.LogbackLock;import org.springframework.util.Assert;/**
* NoSql Appender 基础类
* 子类需要实现generatedKey方法和指定存储类实例
*/abstractpublicclassNoSqlAppender<E>extendsUnsynchronizedAppenderBase<E>{//日志存储protectedILogRepository logRepository;//日志表名protectedString tbname;//日志存储格式protectedString pattern;//日志格式解析器protectedPatternLayout patternLayout;//Nosql操作类protectedObject template;/**
* 日志存储KEY生成
*
* @param event
* @return
*/protectedabstractString generatedKey(E event);/**
* 使用指定的格式生成日志内容数据
*
* @param event
* @return
*/protectedString generatedValue(E event){return patternLayout.doLayout((ILoggingEvent) event);}/**
* 日志表名
*
* @param tbname
*/publicvoid setTbname(String tbname){this.tbname = tbname;}/**
* 日志存储
*
* @param eventObject
*/@Overrideprotectedvoid append(E eventObject){if(!isStarted()){return;}try{String key = generatedKey(eventObject);String value = generatedValue(eventObject);
logRepository.saveLog(key, value);}catch(Exception e){
addError(e.getMessage());}}/**
* 初始化，patternLayout
*/@Overridepublicvoid start(){Assert.notNull(tbname,"tbname not null !");
patternLayout =newPatternLayout();
patternLayout.setPattern(pattern);
patternLayout.setContext(context);
patternLayout.setOutputPatternAsHeader(false);
patternLayout.start();super.start();}@Overridepublicvoid stop(){super.stop();}/**
* 日志格式
*
* @param pattern
*/publicvoid setPattern(String pattern){this.pattern = pattern;}publicvoid setTemplate(Object template){this.template = template;}}ILogRepository
日志存储接口，不同NOSQL数据库提供统一保存方法。
/**
* 日志存储接口
*/public interface ILogRepository<E>{/**
* 保存日志，KEY-VALUE形式
*
* @param key
* @param value
*/publicvoid saveLog(String key,String value);}HbaseLogRepository
Hbase的日志存储实现类
import b2gonline.wap.hbase.HbaseTemplate;import b2gonline.wap.hbase.TableCallback;import b2gonline.wap.logback.ILogRepository;import org.apache.hadoop.hbase.client.HTableInterface;import org.apache.hadoop.hbase.client.Put;import org.apache.hadoop.hbase.util.Bytes;/**
* Hbase 日志存储实现类
*/publicclassHbaseLogRepositoryimplementsILogRepository<HbaseTemplate>{privateString tbname;privateHbaseTemplate hbaseTemplate;//日志列族publicstaticbyte[] CF_INFO =Bytes.toBytes("log");//日志列名privatebyte[] CF_CELL =Bytes.toBytes("data");//日志表名publicvoid setTbname(String tbname){this.tbname = tbname;}publicvoid setHbaseTemplate(HbaseTemplate hbaseTemplate){this.hbaseTemplate = hbaseTemplate;}/**
* 日志数据存储
*
* @param key
* @param value
*/@Overridepublicvoid saveLog(finalString key,String value){finalbyte[] bKey =Bytes.toBytes(key);finalbyte[] bValue =Bytes.toBytes(value);
hbaseTemplate.execute(tbname,newTableCallback<Object>(){@OverridepublicObject doInTable(HTableInterface table)throwsThrowable{Put p =newPut(bKey);
p.add(CF_INFO, CF_CELL, bValue);
table.put(p);returnnull;}});}}HbaseAppender
Hbase的Appender实现类
import b2gonline.wap.hbase.HbaseTemplate;import b2gonline.wap.logback.NoSqlAppender;import ch.qos.logback.classic.spi.ILoggingEvent;import org.apache.hadoop.hbase.HColumnDescriptor;import org.apache.hadoop.hbase.HTableDescriptor;import org.apache.hadoop.hbase.client.HBaseAdmin;import org.springframework.beans.factory.FactoryBean;import org.springframework.beans.factory.InitializingBean;import org.springframework.util.Assert;import java.util.Random;/**
* Hbase Appender 实现类
*/publicclassHbaseAppenderextendsNoSqlAppender<ILoggingEvent>implementsInitializingBean{Random generator =newRandom();privateHBaseAdmin admin;/**
* 初始化，HbaseLogRepository
*/@Overridepublicvoid start(){Assert.notNull(template,"hbaseTemplate not null !");try{
afterPropertiesSet();}catch(Exception e){
e.printStackTrace();}HbaseLogRepository repository =newHbaseLogRepository();
repository.setHbaseTemplate((HbaseTemplate) template);
repository.setTbname(tbname);
logRepository = repository;super.start();}/**
* 生成记录KEY，如果有必要也可以通过patternLayout生成
*
* @param event
* @return
*/@OverrideprotectedString generatedKey(ILoggingEvent event){//使用随机数防止并发生成同名KEYint id = generator.nextInt(9999)+1000;StringBuilder sb =newStringBuilder();
sb.append(event.getLoggerName());
sb.append(event.getLevel());
sb.append(event.getThreadName());
sb.append(event.getTimeStamp());
sb.append("-");
sb.append(id);return sb.toString().toLowerCase().replaceAll(" ","");}/**
* 没有表自动创建
*
* @throws Exception
*/@Overridepublicvoid afterPropertiesSet()throwsException{
admin =newHBaseAdmin(((HbaseTemplate) template).getConfiguration());if(!admin.tableExists(tbname)){HTableDescriptor tableDescriptor =newHTableDescriptor(tbname);HColumnDescriptor columnDescriptor =newHColumnDescriptor(HbaseLogRepository.CF_INFO);
tableDescriptor.addFamily(columnDescriptor);
admin.createTable(tableDescriptor);}}}关于HbaseTemplate
如上所说，Hbase操作类HbaseTemplate提取自SPRING-DATA-HADOOP，参考https://github.com/SpringSource/spring-hadoop/trunk/spring-hadoop-core/src/main/java/org/springframework/data/hadoop/hbase。
提取后会依赖ConfigurationUtils，源码如下：
import org.springframework.util.Assert;import java.util.Enumeration;import java.util.Properties;publicclassConfigurationUtils{publicstaticvoid addProperties(org.apache.hadoop.conf.Configuration configuration,Properties properties){Assert.notNull(configuration,"A non-null configuration is required");if(properties !=null){Enumeration<?> props = properties.propertyNames();while(props.hasMoreElements()){String key = props.nextElement().toString();
configuration.set(key, properties.getProperty(key));}}}}提取后的Hbase连接配置：
<bean id="hbaseConfiguration"class="b2gonline.wap.hbase.HbaseConfigurationFactoryBean"><property name="zkPort" value="2181"/><property name="zkQuorum" value="hadoopmaster,hadoopnode1"/></bean><bean id="hbaseTemplate"class="b2gonline.wap.hbase.HbaseTemplate"><property name="configuration" ref="hbaseConfiguration"/></bean>最后
实现了日志数据统一存储就还得有统一查看的功能，没错，下一步实现！