使用redis存储业务信息，同时也可以存储系统运维信息，比如日志和计数器来收集系统当前的状态信息，挖掘正在使用系统的顾客信息，以及诊断系统问题，发现潜在的问题。当然，系统日志信息及统计信息也可以存储在关系型数据库中，但是存在一个很大的弊端，影响业务性能。
1.使用redis记录日志
熟悉java的朋友，记录日志往往采用的是log4j,sl4j，大多记录载体选择文本文件。如果使用web集群的话，造成日志分散在各个web服务器，搜集有效日志信息，非常麻烦。如果选择数据库保存的话，解决了文件分散情况，但势必对业务造成影响，日志毕竟是个辅助支撑而已，不应该和业务系统相提并论。这时候，redis是一个不错的选择。如果可以的话，可以对log4j扩展，将数据保存到redis中，当然这不是本章的重点。本章重点，主要简单讨论下如何保存日志。
构建一个系统，判断哪些信息需要被记录是一件困难的事情，不同的业务有不同的需求。但一般的日志信息往往关注一下方面。
日志时间，日志内容，服务IP，日志级别，日志发生频率。
1.1redis日志存储设计
记录详情里，可以按要求，增添想要的信息，发生的类名称，处理IP等。
1.2代码
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
|
public
void
logCommon(
Jedis conn, String name, String message, String severity,
int
timeout) {
String commonDest =
"common:"
+ name +
':'
+ severity;
String startKey = commonDest +
":start"
;
long
end = System.currentTimeMillis() + timeout;
while
(System.currentTimeMillis() < end){
conn.watch(startKey);
//当前所处的小时数
String hourStart = ISO_FORMAT.format(
new
Date());
String existing = conn.get(startKey);
Transaction trans = 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);
}
//日志计数器增1
trans.zincrby(commonDest,
1
, message);
//记录最近日志详情
String recentDest =
"recent:"
+ name +
':'
+ severity;
trans.lpush(recentDest, TIMESTAMP.format(
new
Date()) +
' '
+ message);
trans.ltrim(recentDest,
0
,
99
);
List<Object> results = trans.exec();
// null response indicates that the transaction was aborted due to
// the watched key changing.
if
(results ==
null
){
continue
;
}
return
;
}
}
|
2.网站点击量计数器统计
2.1redis计数器存储设计
2.2编码
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
//以秒为单位的精度
public
static
final
int
[] PRECISION =
new
int
[]{
1
,
5
,
60
,
300
,
3600
,
18000
,
86400
};
public
void
updateCounter(Jedis conn, String name,
int
count,
long
now){
Transaction trans = conn.multi();
//每一次更新，都要更新所有精度的计数器
for
(
int
prec : PRECISION) {
long
pnow = (now / prec) * prec;
//当前时间片的开始时间
String hash = String.valueOf(prec) +
':'
+ name;
trans.zadd(
"known:"
,
0
, hash);
trans.hincrBy(
"count:"
+ hash, String.valueOf(pnow), count);
}
trans.exec();
}
public
List<Pair<Integer,Integer>> getCounter(
Jedis conn, String name,
int
precision)
{
String hash = String.valueOf(precision) +
':'
+ name;
Map<String,String> data = conn.hgetAll(
"count:"
+ hash);
ArrayList<Pair<Integer,Integer>> results =
new
ArrayList<Pair<Integer,Integer>>();
for
(Map.Entry<String,String> entry : data.entrySet()) {
results.add(
new
Pair<Integer,Integer>(
Integer.parseInt(entry.getKey()),
Integer.parseInt(entry.getValue())));
}
Collections.sort(results);
return
results;
}
|
2.3清楚旧数据
流程图
代码
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
public class CleanCountersThread
extends Thread
{
private Jedis conn;
private int sampleCount = 100;
private boolean quit;
private long timeOffset;
//
used to mimic a
time
in
the future.
public CleanCountersThread(int sampleCount, long timeOffset){
this.conn = new Jedis(
"192.168.163.156"
);
this.conn.
select
(15);
this.sampleCount = sampleCount;
this.timeOffset = timeOffset;
}
public void quit(){
quit =
true
;
}
public void run(){
int passes = 0;
while
(!quit){
long start = System.currentTimeMillis() + timeOffset;
int index = 0;
while
(index < conn.zcard(
"known:"
)){
Set<String> hashSet = conn.zrange(
"known:"
, index, index);
index++;
if
(hashSet.size() == 0) {
break
;
}
String
hash
= hashSet.iterator().next();
int prec = Integer.parseInt(
hash
.substring(0,
hash
.indexOf(
':'
)));
int bprec = (int)Math.floor(prec / 60);
if
(bprec == 0){
bprec = 1;
}
if
((passes % bprec) != 0){
continue
;
}
String hkey =
"count:"
+
hash
;
String cutoff = String.valueOf(
((System.currentTimeMillis() + timeOffset) / 1000) - sampleCount * prec);
ArrayList<String> samples = new ArrayList<String>(conn.hkeys(hkey));
Collections.
sort
(samples);
int remove = bisectRight(samples, cutoff);
if
(remove != 0){
conn.hdel(hkey, samples.subList(0, remove).toArray(new String[0]));
if
(remove == samples.size()){
conn.
watch
(hkey);
if
(conn.hlen(hkey) == 0) {
Transaction trans = conn.multi();
trans.zrem(
"known:"
,
hash
);
trans.
exec
();
index--;
}
else
{
conn.unwatch();
}
}
}
}
passes++;
long duration = Math.min(
(System.currentTimeMillis() + timeOffset) - start + 1000, 60000);
try {
sleep
(Math.max(60000 - duration, 1000));
}catch(InterruptedException ie){
Thread.currentThread().interrupt();
}
}
}
//
mimic python's bisect.bisect_right
public int bisectRight(List<String> values, String key) {
int index = Collections.binarySearch(values, key);
return
index < 0 ? Math.abs(index) - 1 : index + 1;
}
}
|
3.使用redis统计数据
上面提到的计数器，是最简单的统计数据。除了计数器（count(*)）,还是最大值(max),最小值(min).
设计
stats:模块(页面)名称:指标名称
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
|
public
List<Object> updateStats(Jedis conn, String context, String type,
double
value){
int
timeout =
5000
;
String destination =
"stats:"
+ context +
':'
+ type;
String startKey = destination +
":start"
;
long
end = System.currentTimeMillis() + timeout;
while
(System.currentTimeMillis() < end){
conn.watch(startKey);
String hourStart = ISO_FORMAT.format(
new
Date());
String existing = conn.get(startKey);
Transaction trans = 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提供的最大值，最小值计算
String tkey1 = UUID.randomUUID().toString();
String tkey2 = UUID.randomUUID().toString();
trans.zadd(tkey1, value,
"min"
);
trans.zadd(tkey2, value,
"max"
);
trans.zunionstore(
destination,
new
ZParams().aggregate(ZParams.Aggregate.MIN),
destination, tkey1);
trans.zunionstore(
destination,
new
ZParams().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
;
}
return
results.subList(results.size() -
3
, results.size());
}
return
null
;
}
|
需要注意的使用redis自带的最大值最小值，计算,所以创建了2个临时有序集合。其他的逻辑参照日志相关部分。
本文转自 randy_shandong 51CTO博客，原文链接：http://blog.51cto.com/dba10g/1853096，如需转载请自行联系原作者
