Twitter Storm源代码分析之TimeCacheMap

TimeCacheMap是Twitter Storm里面一个类， Storm使用它来保存那些最近活跃的对象，并且可以自动删除那些已经过期的对象。这个类设计的很巧妙， 我们来看一下。

TimeCacheMap里面的数据是保存在内部变量_bucket里面的:

帮助

1	private LinkedList<HashMap<K, V>> _buckets;

在这点上跟ConcurrentHashMap有点类似， ConcurrentHashMap是利用多个bucket来缩小锁的粒度, 从而实现高并发的读写。而TimeCacheMap则是利用多个bucket来使得数据清理线程占用锁的时间最小。

首先来看看TimeCacheMap的构造函数， 它的构造函数首先是生成numBuckets个空的HashMap：

帮助

1 2 3 4

_buckets = new LinkedList<HashMap<K, V>>(); for(int i=0; i<numBuckets; i++) {     _buckets.add(new HashMap<K, V>()); }

然后就是最关键的清理线程部分，TimeCacheMap使用一个单独的线程来清理那些过期的数据:

帮助

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

final long expirationMillis = expirationSecs * 1000L; final long sleepTime = expirationMillis/(numBuckets-1); _cleaner = new Thread(new Runnable() {     public void run() {         try {             while(true) {                 Map<K, V> dead = null;                 //                 Time.sleep(sleepTime);                 synchronized(_lock) {                   dead = _buckets.removeLast();                   _buckets.addFirst(                              new HashMap<K, V>());                 }                 if(_callback!=null) {                     for(Entry<K, V> entry:                                 dead.entrySet()) {                       _callback.expire(entry.getKey(),                                     entry.getValue());                     }                 }             }         } catch (InterruptedException ex) {           }     } }); _cleaner.setDaemon(true); _cleaner.start();

这个线程每隔 expirationSecs / (numBuckets - 1) 秒钟的时间去把最后一个bucket里面的数据全部都删除掉 — 这些被删除掉的数据其实就是过期的数据。(为什么不是每隔expirationSecs就来删除一次呢？我们下面会说)。这里值得注意的是：正是因为这种分成多个桶的机制， 清理线程对于_lock的占用时间极短。只要把最后一个bucket从_buckets解下，并且向头上面添加一个新的bucket就好了：

帮助

1 2 3 4

synchronized(_lock) {     dead = _buckets.removeLast();     _buckets.addFirst(new HashMap<K, V>()); }

如果不是这种机制的话， 那我能想到的最傻的办法可能就是给条数据一个过期时间字段， 然后清理线程就要遍历每条数据来检查数据是否过期了。那显然要HOLD住这个锁很长时间了。

同时对于每条过期的数据TimeCacheMap会执行我们的callback函数：

帮助

1 2 3 4 5 6

if(_callback!=null) {      for(Entry<K, V> entry: dead.entrySet()) {           _callback.expire(entry.getKey(),                            entry.getValue());      } }

大致机制就是这样，那么我们现在回过头来看看前面的那个问题： 为什么这个清理线程是每隔expirationSecs / (numBuckets - 1) 秒的时间来检查，这样对吗？TimeCacheMap的内部有多个桶， 当你向这个TimeCacheMap里面添加数据的时候，数据总是添加到第一个桶里面去的。

帮助

01 02 03 04 05 06 07 08 09 10 11 12

public void put(K key, V value) {     synchronized(_lock) {         Iterator<HashMap<K, V>> it =                          _buckets.iterator();         HashMap<K, V> bucket = it.next();         bucket.put(key, value);         while(it.hasNext()) {             bucket = it.next();             bucket.remove(key);         }     } }

我们看个例子就明白了，假设 numBuckets = 3, expirationSecs = 2。
我们先往里面填一条数据{1: 1}， 这条数据被加到第一个桶里面去， 现在TimeCacheMap的状态是:

帮助

1	[{1:1}, {}, {}]

过了1秒钟之后(expirationSecs / (numBuckets - 1) = 2 / (3 - 1) = 1)。清理线程干掉最后一个HashMap，并且在头上添加一个新的空HashMap, 现在TimeCacheMap的状态是:

帮助

1	[{}, {1:1}, {}]

再过了一秒钟， 同上， TimeCacheMap的状态会变成：

帮助

1	[{}, {}, {1:1}]

再过一秒钟， 现在{1:1}是最后一个TimeCacheMap了，就被干掉了。
所以从{1:1}被加入到这个TimeCacheMap到被干掉一共用了3秒，其实这个3秒就等于

帮助

1	3 = expirationSecs * ( 1 + 1 / (numBuckets - 1))

它的注释里面也提到了这一点

Expires keys that have not been updated in the configured number of seconds.
The algorithm used will take between expirationSecs and
expirationSecs * (1 + 1 / (numBuckets-1)) to actually expire the message.

那为什么说时间是expirationSecs到expirationSecs * (1 + 1 / (numBuckets-1))之间呢？因为线程调度的不确定性。

转自淘宝