storm0.8.1以后的RotatingMap完全可以独立于storm用来实现hashmap的key超时删除，并调用回调函数

RotatingMap.java：

import java.util.HashMap;

import java.util.Iterator;

import java.util.LinkedList;

import java.util.Map;

import java.util.Map.Entry;

/**

 * 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.

 *

 * get, put, remove, containsKey, and size take O(numBuckets) time to run.

 *

 * The advantage of this design is that the expiration thread only locks the object

 * for O(1) time, meaning the object is essentially always available for gets/puts.

 */

public class RotatingMap<K, V> {

    //this default ensures things expire at most 50% past the expiration time

    private static final int DEFAULT_NUM_BUCKETS = 3;

    public static interface ExpiredCallback<K, V> {

        public void expire(K key, V val);

    }

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

    private ExpiredCallback _callback;

    public RotatingMap(int numBuckets, ExpiredCallback<K, V> callback) {

        if(numBuckets<2) {

            throw new IllegalArgumentException("numBuckets must be >= 2");

        }

        _buckets = new LinkedList<HashMap<K, V>>();

        for(int i=0; i<numBuckets; i++) {

            _buckets.add(new HashMap<K, V>());

        }

        _callback = callback;

    }

    public RotatingMap(ExpiredCallback<K, V> callback) {

        this(DEFAULT_NUM_BUCKETS, callback);

    }

    public RotatingMap(int numBuckets) {

        this(numBuckets, null);

    }   

    public Map<K, V> rotate() {

        Map<K, V> 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());

            }

        }

        return dead;

    }

    public boolean containsKey(K key) {

        for(HashMap<K, V> bucket: _buckets) {

            if(bucket.containsKey(key)) {

                return true;

            }

        }

        return false;

    }

    public V get(K key) {

        for(HashMap<K, V> bucket: _buckets) {

            if(bucket.containsKey(key)) {

                return bucket.get(key);

            }

        }

        return null;

    }

    public void put(K key, V value) {

        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);

        }

    }

    public Object remove(K key) {

        for(HashMap<K, V> bucket: _buckets) {

            if(bucket.containsKey(key)) {

                return bucket.remove(key);

            }

        }

        return null;

    }

    public int size() {

        int size = 0;

        for(HashMap<K, V> bucket: _buckets) {

            size+=bucket.size();

        }

        return size;

    }

}

EventHandler.java

public class EventHandler<k, v> implements RotatingMap.ExpiredCallback<k, v>

{

 @Override

 public void expire(Object key, Object val)

 {

  System.out.println("key=" + key + ",val=" + val);

 }

}

RotatingMapStarter.java：

import java.sql.Date;

import java.text.SimpleDateFormat;

public class RotatingMapStarter

{

 RotatingMap<String, String> m_rotatingMap = null;

 RotatingMap.ExpiredCallback<String, String> m_eventHandler = null;

 long m_lastRotate = System.currentTimeMillis();

 long m_rotateTime;

 @SuppressWarnings(

 { "unchecked", "rawtypes" })

 public RotatingMapStarter(int n, int rotateTime) // rotateTime :rotate for

              // second

 {

  m_eventHandler = new EventHandler<String, String>();

  m_rotatingMap = new RotatingMap<String, String>(4, m_eventHandler);

  m_lastRotate = System.currentTimeMillis();

  m_rotateTime = 1000L * rotateTime; // millisecond

 }

 public RotatingMap<String, String> getM_rotatingMap()

 {

  return m_rotatingMap;

 }

 public void setM_rotatingMap(RotatingMap<String, String> m_rotatingMap)

 {

  this.m_rotatingMap = m_rotatingMap;

 }

 public void StartConnMonitor()

 {

  Thread thread = new Thread("Server Monitor")

  {

   @SuppressWarnings("static-access")

   public void run()

   {

    while (true)

    {

     try

     {

      Thread.currentThread().sleep(1000);

     }

     catch (InterruptedException e)

     {

      e.printStackTrace();

     }

     long now = System.currentTimeMillis();

     if (now - m_lastRotate > m_rotateTime)

     {

      m_rotatingMap.rotate();

      m_lastRotate = now;

     }

     else

     {

      //System.out.println(now - m_lastRotate);

      SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式

      System.out.println(df.format(new Date(now)));// new Date()为获取当前系统时间

     }

    }

   }

  };

  thread.start();

 }

 /**

  * @param args

  */

 public static void main(String[] args)

 {

  RotatingMapStarter rotatingMapStarter = new RotatingMapStarter(4, 10);

  String value = "001";

  String key = "value";

  rotatingMapStarter.getM_rotatingMap().put(key, value);

  rotatingMapStarter.StartConnMonitor();

 }

}