监控客户端设计-记录-输出部分

现在正在做一个监控的工作，看似简单，但如果多想想还是有很多技巧和问题可以探讨的

 

需求场景：

 

1. 客户端应该是一个静态类
2. 客户端调用该类的记录信息API频度可能很大
3. 该类记录信息的API应该是异步的，保证主体性能
4. 能够提供多种记录的方式
5. 能够记录多种数据方式

需求其实可以分为两部分：

1. 对于需要记录的信息最终可能有多种形式来表达
2. 使用什么方式（媒体）来表达，比如log，socket，或者消息系统

 

所以设计的时候，在最开始用Recorder和Out两个接口来表达。

 

其中Recorder中定义以下方法

 

/**
 * 信息记录者借口
 * @author dongxuan.lb
 *
 */
public interface Recorder {
	/**
	 * 记录信息
	 * @param keys
	 */
	void record(String... keys);
	
	/**
	 * 该Recorder的名称
	 * @return
	 */
	String getName();
	
	void setName(String name);
	
	/**
	 * 将记录输出
	 * 同时清理内部数据
	 * 内部使用Out接口
	 */
	void free();
}

 

 Out接口很简单：

 

/**
 * 记录输出接口
 * @author dongxuan.lb
 *
 */
public interface Out {
	void output(String info);
}

 

 复杂的输出由具体的子类来完成。

 

到这里，这两个接口是完全解耦的，彼此没有什么关系。

 

========================================================================

到这里，可以开始设计具体的Recorder类了。（不同的业务需求可以自己实现Recorder）

 

这里需要具体化的需求：

 

 写道

主体会接收到一个字符窜数组，不定长度

该API能够在一定时期后输出出现过的key以及该key的次数

比如A.record("a","a","b")

可能的结果是

a=2|b=1

当然这里record中的字符串长度不会非常大，通常小于100个，但是调用该api的频度非常大，可能在1000至10000

 

 监控方面的设计，在使用API记录是不能同步阻塞住 主体业务，所以使用线程方式进行异步。

 

 

在这里使用了ThreadPoolExecutor作为作为线程池。

 

 

protected ThreadPoolExecutor recorderPool = new ThreadPoolExecutor(
			coreSize, coreSize * 2, 30, TimeUnit.SECONDS, 
			new ArrayBlockingQueue<Runnable>(60), 
			new RecorderThreadFactory("recorderPool"), 
			new ThreadPoolExecutor.CallerRunsPolicy());

 这里需要注意的是coreSize和maxPoolSize的设置，当然这里也是可以调优的地方。

 

当然使用ThreadPoolExecutor最主要的是RejectedExecutionHandler的具体实现可以高度自由化

 

当核心处理不过来的时候可以选择自己实现的RejectedExecutionHandler，以满足业务需求

 

AbstractKeyCountRecorder中代码

 

private Integer coreSize = Runtime.getRuntime().availableProcessors();
	/*
	 * 记录者线程池
	 */
	protected ThreadPoolExecutor recorderPool = new ThreadPoolExecutor(
			coreSize, coreSize * 2, 30, TimeUnit.SECONDS, 
			new ArrayBlockingQueue<Runnable>(60), 
			new RecorderThreadFactory("recorderPool"), 
			new ThreadPoolExecutor.CallerRunsPolicy());
	
	protected abstract Runnable getWorker(String... key);
	
	@Override
	public void record(String... keys) {
		recorderPool.execute(getWorker(keys));
	}

 

 

========================================================================

回到具体需求上来。

 

首先为了记录key的次数，需要一个map来作为记录存数，其中map的K中存储key的名称，map的V中存储key的出现次数。

 

但这里有两个问题：

 

1. 使用hashMap线程不安全，get和put方法均有可能出现并发问题，导致记录错误
2. map中的V存在自加操作，在操作也会有线程问题

事实证明，

map的同步影响的是性能问题，而V得类型选择则非常影响最终结果的正确与否（当然背后还是同步问题）

========================================================================

map的同步选择：

1. 选择hashtable
2. 使用Collections.synchronizedMap()方式
3. 使用ConcurrentHashMap
4. 自己手动维护同步

在这里，不想过多介绍这个问题，不过比较奇怪的是，我测试下来，Collections.synchronizedMap()方式比较耗性能

这里有篇文章和我的测试大致一样：http://blog.csdn.net/java2000_net/archive/2008/11/25/3373181.aspx

最终为了方便，直接选择ConcurrentHashMap

========================================================================

这里需要重点指出的是，map中V的选型，通常就使用Integer就可以了，但是在这里，虽然map的put，get方法已经同步，但是当多个线程去取同一个V的值时可能看到的是一样的，这样当自加后在put进去，就丢掉了几次记录。

所以如果使用Integer作为计数器类，结果必然小了很多。

那么自然想到的便是AtomInteger类了。内部使用的机制比较复杂，在《java Thread》那本书上有提到过。

 写道

private Map<String, AtomicInteger> hitMap = new ConcurrentHashMap<String, AtomicInteger>(1000);

 ========================================================================

之前说过，Recorder和Out很好的解耦了，到了具体的Recorder中，由传入的out负责具体的输出部分

外部注入out实体

public LevelKeyCountRecorder(Out out) {
		this();
		this.out = out;
	}

 当需要输出时调用Out的output方法

@Override
	public void free() {
		if(!hitMap.isEmpty()) {
			StringBuilder value = new StringBuilder();
			for (final Entry<String, AtomicInteger> entry : hitMap.entrySet()) {
				value.append(
						entry.getKey()
						+ "=" 
						+ entry.getValue() 
						+ "|");
			}
			out.output(value.toString());
			hitMap.clear();
		}
	}

 =======================================================================

最后 ，使用日志输出结果

Logger logger = Logger.getLogger(LevelKeyCountRecorder.class);
		Out logOut = new LogOut(logger);
		Recorder levelRecorder = new LevelKeyCountRecorder(logOut)

 

评论

1 楼 kkqqcom 2011-01-16  

我也在做类似的事情。
整理整理，一起交流。

我的场景要求高稳定性和吞吐量，并发不大。