使用redis存储业务信息，同时也可以存储系统运维信息，比如日志和计数器来收集系统当前的状态信息，挖掘正在使用系统的顾客信息，以及诊断系统问题，发现潜在的问题。当然，系统日志信息及统计信息也可以存储在关系型数据库中，但是存在一个很大的弊端，影响业务性能。

1.使用redis记录日志

熟悉java的朋友，记录日志往往采用的是log4j,sl4j，大多记录载体选择文本文件。如果使用web集群的话，造成日志分散在各个web服务器，搜集有效日志信息，非常麻烦。如果选择数据库保存的话，解决了文件分散情况，但势必对业务造成影响，日志毕竟是个辅助支撑而已，不应该和业务系统相提并论。这时候，redis是一个不错的选择。如果可以的话，可以对log4j扩展，将数据保存到redis中，当然这不是本章的重点。本章重点，主要简单讨论下如何保存日志。

构建一个系统，判断哪些信息需要被记录是一件困难的事情，不同的业务有不同的需求。但一般的日志信息往往关注一下方面。

日志时间，日志内容，服务IP，日志级别，日志发生频率。

1.1redis日志存储设计

记录详情里，可以按要求，增添想要的信息，发生的类名称，处理IP等。

1.2代码

publicvoidlogCommon(Jedisconn,Stringname,Stringmessage,Stringseverity,inttimeout){StringcommonDest="common:"+name+':'+severity;StringstartKey=commonDest+":start";longend=System.currentTimeMillis()+timeout;while(System.currentTimeMillis()<end){conn.watch(startKey);//当前所处的小时数StringhourStart=ISO_FORMAT.format(newDate());Stringexisting=conn.get(startKey);Transactiontrans=conn.multi();//如果记录的是上一个小时的日志if(existing!=null&&COLLATOR.compare(existing,hourStart)<0){trans.rename(commonDest,commonDest+":last");trans.rename(startKey,commonDest+":pstart");trans.set(startKey,hourStart);}else{trans.set(startKey,hourStart);}//日志计数器增1trans.zincrby(commonDest,1,message);//记录最近日志详情StringrecentDest="recent:"+name+':'+severity;trans.lpush(recentDest,TIMESTAMP.format(newDate())+''+message);trans.ltrim(recentDest,0,99);List<Object>results=trans.exec();//nullresponseindicatesthatthetransactionwasaborteddueto//thewatchedkeychanging.if(results==null){continue;}return;}}

2.网站点击量计数器统计

2.1redis计数器存储设计

2.2编码

//以秒为单位的精度publicstaticfinalint[]PRECISION=newint[]{1,5,60,300,3600,18000,86400};publicvoidupdateCounter(Jedisconn,Stringname,intcount,longnow){Transactiontrans=conn.multi();//每一次更新，都要更新所有精度的计数器for(intprec:PRECISION){longpnow=(now/prec)*prec;//当前时间片的开始时间Stringhash=String.valueOf(prec)+':'+name;trans.zadd("known:",0,hash);trans.hincrBy("count:"+hash,String.valueOf(pnow),count);}trans.exec();}publicList<Pair<Integer,Integer>>getCounter(Jedisconn,Stringname,intprecision){Stringhash=String.valueOf(precision)+':'+name;Map<String,String>data=conn.hgetAll("count:"+hash);ArrayList<Pair<Integer,Integer>>results=newArrayList<Pair<Integer,Integer>>();for(Map.Entry<String,String>entry:data.entrySet()){results.add(newPair<Integer,Integer>(Integer.parseInt(entry.getKey()),Integer.parseInt(entry.getValue())));}Collections.sort(results);returnresults;}

2.3清楚旧数据

流程图

代码

publicclassCleanCountersThreadextendsThread{privateJedisconn;privateintsampleCount=100;privatebooleanquit;privatelongtimeOffset;//usedtomimicatimeinthefuture.publicCleanCountersThread(intsampleCount,longtimeOffset){this.conn=newJedis("192.168.163.156");this.conn.select(15);this.sampleCount=sampleCount;this.timeOffset=timeOffset;}publicvoidquit(){quit=true;}publicvoidrun(){intpasses=0;while(!quit){longstart=System.currentTimeMillis()+timeOffset;intindex=0;while(index<conn.zcard("known:")){Set<String>hashSet=conn.zrange("known:",index,index);index++;if(hashSet.size()==0){break;}Stringhash=hashSet.iterator().next();intprec=Integer.parseInt(hash.substring(0,hash.indexOf(':')));intbprec=(int)Math.floor(prec/60);if(bprec==0){bprec=1;}if((passes%bprec)!=0){continue;}Stringhkey="count:"+hash;Stringcutoff=String.valueOf(((System.currentTimeMillis()+timeOffset)/1000)-sampleCount*prec);ArrayList<String>samples=newArrayList<String>(conn.hkeys(hkey));Collections.sort(samples);intremove=bisectRight(samples,cutoff);if(remove!=0){conn.hdel(hkey,samples.subList(0,remove).toArray(newString[0]));if(remove==samples.size()){conn.watch(hkey);if(conn.hlen(hkey)==0){Transactiontrans=conn.multi();trans.zrem("known:",hash);trans.exec();index--;}else{conn.unwatch();}}}}passes++;longduration=Math.min((System.currentTimeMillis()+timeOffset)-start+1000,60000);try{sleep(Math.max(60000-duration,1000));}catch(InterruptedExceptionie){Thread.currentThread().interrupt();}}}//mimicpython'sbisect.bisect_rightpublicintbisectRight(List<String>values,Stringkey){intindex=Collections.binarySearch(values,key);returnindex<0?Math.abs(index)-1:index+1;}}

3.使用redis统计数据

上面提到的计数器，是最简单的统计数据。除了计数器（count(*)）,还是最大值(max),最小值(min).

设计

stats:模块(页面)名称:指标名称

publicList<Object>updateStats(Jedisconn,Stringcontext,Stringtype,doublevalue){inttimeout=5000;Stringdestination="stats:"+context+':'+type;StringstartKey=destination+":start";longend=System.currentTimeMillis()+timeout;while(System.currentTimeMillis()<end){conn.watch(startKey);StringhourStart=ISO_FORMAT.format(newDate());Stringexisting=conn.get(startKey);Transactiontrans=conn.multi();if(existing!=null&&COLLATOR.compare(existing,hourStart)<0){trans.rename(destination,destination+":last");trans.rename(startKey,destination+":pstart");trans.set(startKey,hourStart);}//借助redis提供的最大值，最小值计算Stringtkey1=UUID.randomUUID().toString();Stringtkey2=UUID.randomUUID().toString();trans.zadd(tkey1,value,"min");trans.zadd(tkey2,value,"max");trans.zunionstore(destination,newZParams().aggregate(ZParams.Aggregate.MIN),destination,tkey1);trans.zunionstore(destination,newZParams().aggregate(ZParams.Aggregate.MAX),destination,tkey2);trans.del(tkey1,tkey2);trans.zincrby(destination,1,"count");trans.zincrby(destination,value,"sum");trans.zincrby(destination,value*value,"sumsq");List<Object>results=trans.exec();if(results==null){continue;}returnresults.subList(results.size()-3,results.size());}returnnull;}

需要注意的使用redis自带的最大值最小值，计算,所以创建了2个临时有序集合。其他的逻辑参照日志相关部分。

参考内容

《redis in action》