[面试题]java高性能的多计数器

题目要求

需要实现一个并发安全的多计数器，可以记录不同数字出现的次数，简单而言就是类似于 Map<Integer, AtomicInteger> 的数据结构，但该数据结构所占用的内存较大：一个 Node<Integer, AtomicInteger> 对象就需要占用 96Byte，如果算上 Node[] 则该数据结构整体的内存占用将是非常可观的。

综上所述，我们需要实现一个低内存开销、并发安全的多计数器 MultiAtomicInteger:

public interface MultiAtomicInteger {

    /**
     * 原子的将 key 对应的值加上 cnt, 并返回更新后的值
     *
     * @param key : 非负整数，取值范围为 [0 ~ Integer.MAX_VALUE]
     * @param cnt : 增量（正数）
     * @return : 更新后的值
     */
    int addAndGet(int key, int cnt);

    /**
     * 返回 key 对应的值
     *
     * @param key : 非负整数，取值范围为 [0 ~ Integer.MAX_VALUE]
     * @return : key 对应的值
     */
    int get(int key);

    /**
     * 返回当前一共有多少 key
     */
    int size();
}

为了降低编程难度，只需要实现一个无需扩容的 MultiAtomicInteger 即可，该类的构造函数需要指定最大容量。

答案：

public class MyMultiAtomicInteger implements MultiAtomicInteger {

    /**
     * lock集合
     */
    private final Object[] locks;

    /**
     * [hash]
     */
    private final long[][] map;

    /**
     * 0有特殊含义，单独用一个变量记
     */
    private final AtomicInteger zeroNum = new AtomicInteger();
    /**
     * 标记0是否放入过
     */
    private final AtomicBoolean existsZero = new AtomicBoolean();

    private final AtomicInteger size = new AtomicInteger();

    public ZslMultiAtomicInteger(int capacity) {
        // 系数越大，吞吐量越高，内存也越多
//        capacity = (int) Math.max(1, capacity * 0.35);
        capacity = (int) Math.max(1, capacity * 0.4);

        // 理论上锁数量越多，吞吐量越高，但是太多的话，占用内存也多
        int lockNum = Math.min(128, capacity);
        locks = new Object[lockNum];
        for (int i = 0; i < locks.length; i++) {
            locks[i] = new Object();
        }

        map = new long[capacity][];
    }

    @Override
    public int addAndGet(int key, int cnt) {
        if (key == 0) {
            cnt = zeroNum.addAndGet(cnt);
            boolean exists = existsZero.getAndSet(true);
            if (!exists) {
                size.incrementAndGet();
            }
            return cnt;
        }

        int hash = hash(key);
        int lockIndex = hash % locks.length;
        synchronized (locks[lockIndex]) {
            long[] arr = map[hash];
            if (arr == null) {
                arr = map[hash] = new long[2]; // 初始化链表大小2
            }

            for (int i = 0; i < arr.length; i++) {
                if (arr[i] == 0) {
                    arr[i] = createNode(key, cnt);
                    size.incrementAndGet();
                    return cnt;
                }

                if (key == getKey(arr[i])) {
                    arr[i] += cnt;
                    return getValue(arr[i]);
                }
            }

            int oldLen = arr.length;
            arr = map[hash] = Arrays.copyOf(arr, Math.min(arr.length << 1, Integer.MAX_VALUE));
            arr[oldLen] = createNode(key, cnt);
            size.incrementAndGet();
            return cnt;
        }
    }

    @Override
    public int get(int key) {
        if (key == 0) {
            return zeroNum.get();
        }

        int hash = hash(key);
        long[] arr = map[hash];
        if (arr == null) {
            return 0;
        }

        for (long node : arr) {
            if (node == 0) {
                return 0;
            }

            if (key == getKey(node)) {
                return getValue(node);
            }
        }

        return 0;
    }

    @Override
    public int size() {
        return size.get();
    }

    private int hash(int key) {
        return key % map.length;
    }

    private long createNode(int key, int value) {
        return (long) key << 32 | value;
    }

    private int getKey(long node) {
        return (int) (node >> 32);
    }

    private int getValue(long node) {
        return (int) (node);
    }
}

我遇到问题：

• 如何更省空间的保存<Integer,Integer>呢，如果用数组，那如何根据Key->数组下标呢，如果还做一个map映射，那就没有空间的意义了？
• 如何设计一个key维度的锁？

解读：（例如1000个key，capacity=1000）

• 存储结构：使用long[][] map二维数组存储，根据key%map.length得到二维数组的行下标。一维数组中单个元素是long类型，高位存int类型的Key，低位存int类型Value。一维数组中存储的是所有行下标的相同的元素。例如（1和401的对应的下标都是1，这两个数的计算会放到map[1]中，形如[47218742,219474827]，第一个元素表示一个k-v，第二个元素表示k-v）
• 创建n个锁：locks = new Object[lockNum];
• 加锁：synchronized (locks[lockIndex]){ }
• 锁维度：锁的个数是1~二维数组的行数。