动态分类计数器

昨天的工作遇到一个需求：要求根据用户 ID（Long 型）记录他的访问某个页面的次数。并且在所有用户的累积计数达到某个值后输出、清空并重新计数。这个记数有个特点，某些用户的访问次数会异乎寻常地多。

因为这个记录只是在高访问量的时候做，所以对程序的并发度要求比较高。我们知道，高访问量下数字型对象的装箱拆箱会极大影响效率；在高并发下，锁竞争也会极大影响效率。

对于装箱拆箱的问题，我们可以在一开始的时候用 new Long(0) 之类的来初始化计数器，但是之后的自增计数器的操作还是会先进行拆箱，在栈里运算后再进行装箱。并发的问题我们看起来可以用 ConcurrentHashMap 来解决。但是对于这个需求来说，我们需要在自增计数器之前先把之前的值取出来，所以就需要把 ConcurrentHashMap 给锁住，这样就失去了使用 ConcurrentHashMap 的意义，使得它褪化成了 SynchronizedMap。当然也可以在 CHM 中放入一个自己封装的对象，其中包含计数器，使用的时候给这个封装过的对象上锁。

我个人的想法是使用自定义的 HashMap（不实现 Map 接口），支持并发地自增计数器，同时避免装箱拆箱的问题。最好还能进行分段加锁。因为分段加锁比较复杂，所以我目前没有实现。下面的程序实现了在避免装箱拆箱的前提下并发自增计数器。但不支持动态扩容，不支持序列化。使用的 hash 算法和数据结构抄袭了 JDK 6 的 HashMap。所以如果你感兴趣自己调试的话，你会发现 java.util.HashMap 的 table 数组的内容与此类的 table 数组的内容分布完全一致（前提是两者的数据一样，容量一样）。

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.sun.istack.internal.Nullable;

/**
 * 并发计数器哈希 Map（没有实现 {@link Map} 接口）。此类适合于进行动态分类统计。因为不支持动态扩容，所以要求计数达到某个值后
 * 手动清空此 Map 并重新计数（通常客户代码此时应将 Map 清空前的内容写入日志）。如果强行在小容量的时候塞入大量
 * 数据，此 Map 不会被撑爆，但是性能会急剧下降。
 * <p>
 * 性能提示：对于某一部分键的计数器自增操作特别多时此类的表现最佳。如果每个键的计数都很少而键却很多时此类的性能最差。
 * 重置操作会锁住整个实例，使得任何计数器的读取和写入操作都被阻塞。
 * <p>
 * 此类的所有公共接口都是线程安全的。
 * <p>
 * 此类不支持序列化和反序列化。
 * <p>
 * 类的关键实现拷贝了 JDK 6 的 HashMap。
 * 
 * @author ydong
 */
public final class ConcurrentCounterHashMap {
	
	/**
	 * 构造一个新实例。capacity 参数指定了容量。此容量一但指定就不会更改。此实例也不会动态扩容。
	 * 强行塞入过多数据会导致性能下降。
	 * 
	 * @param capacity 容量
	 */
	public ConcurrentCounterHashMap(final int capacity) {
		if (capacity <= 0) throw new IllegalArgumentException();
		if (capacity > 2 << 30) capacity = 2 << 30;
		int _capacity = 1;
		while (_capacity < capacity) _capacity <<= 1;
		
		table = new Entry[_capacity];
	}
	
	/**
	 * 对于指定的 key，其相应的计数器自增一。此方法的副作用是会导致调用 {@link #getCount()} 的值加一。
	 * 
	 * @param key 需要自增一的计数器的键
	 * @return 计数器自增一前的值
	 */
	public long incrementAndGet(final long key) {
		try {
			tableRLock.lock();
			
			final int i = indexFor(hash(hash(key)), table.length);
			
			try {
				for (Entry ent = table[i]; ent != null; ent = ent.next) {
					if (ent.key == key) {
						return ent.incrementAndGet();
					}
				}
			} finally {
				tableRLock.unlock();
			}
			
			tableWLock.lock();
			try {
				table[i] = new Entry(key, 1, table[i]);
				return 0;
			} finally {
				tableWLock.unlock();
			}
		} finally {
			count.incrementAndGet();
		}
	}
	
	/**
	 * 获取与指定的 key 关联的计数器的当前值。
	 * 
	 * @param key 要获取的计数器的键
	 * @return 与指定的 key 关联的计数器的当前值
	 */
	public long get(final long key) {
		tableRLock.lock();
		try {
			final int i = indexFor(hash(hash(key)), table.length);
			
			for (Entry ent = table[i]; ent != null; ent = ent.next) {
				if (ent.key == key) return ent.getCount();
			}
			
			return 0;
		} finally {
			tableRLock.unlock();
		}
	}
	
	/**
	 * 获取所有计数器自增次数，即 {@link #incrementAndGet(long)} 的调用次数。
	 * 此值会在 {@link #reset()} 方法调用后重置为 0。
	 * 
	 * @return 所有计数器自增次数
	 */
	public int getCount() {
		return count.get();
	}
	
	/**
	 * 将当前计数器哈希 Map 重置。所有的计数器及与其关联的键都被销毁。实例恢复到构造函数刚被调用后的状态。
	 * 方法返回此实例被重置前的数据，以 {@link Map} 的形式给出。返回的 {@code Map} 不会受到本实例
	 * 后续操作的影响，在不加锁的情况下可以安全使用。
	 * <p>
	 * 注意此方法会将整个实例锁住，任何计数器自增操作和读取操作都会阻塞。
	 * 
	 * @return 此实例被重置前的数据，以 {@link Map} 的形式给出
	 */
	public Map<Long, Long> dumpAndClear() {
		tableWLock.lock();
		try {
			final Map<Long, Long> ret = new HashMap<Long, Long>();
			
			for (int i = 0; i < table.length; i++) {
				final Entry ent = table[i];
				if (ent != null) {
					ret.put(ent.getKey(), ent.getCount());
					table[i] = null;
				}
			}
			
			return ret;
		} finally {
			tableWLock.unlock();
			count.set(0);
		}
	}
	
	private final Entry[] table;
	private final ReadWriteLock tableLock = new ReentrantReadWriteLock();
	private final Lock tableRLock = tableLock.readLock();
	private final Lock tableWLock = tableLock.writeLock();
	private final AtomicInteger count = new AtomicInteger(0);
	
	private static int hash(final long key) { return (int)(key ^ (key >>> 32)); }
	
	private static int hash(int h) {
        h ^= (h >>> 20) ^ (h >>> 12);
        return h ^ (h >>> 7) ^ (h >>> 4);
    }
	
	private static int indexFor(final int hash, final int length) { return hash & (length - 1); }
	
	private static final class Entry {
		
		public Entry(final long key, final long count, @Nullable final Entry next) {
			this.key = key;
			this.count.set(count);
			this.next = next;
		}
		
		public long getKey() { return key; }
		
		public long getCount() { return count.get(); }
		
		public long incrementAndGet() { return count.incrementAndGet(); }
		
		@Override
		public String toString() {
			return String.format("%d=%d", key, count);
		}

		final long key;
		final AtomicLong count = new AtomicLong(0L);
		@Nullable final Entry next;
		
	}

}