Skip to content

javalib jctools map

wangkaiwu edited this page Feb 27, 2023 · 2 revisions

NonBlockingHashMap

构造

构造
public class NonBlockingHashMap<TypeK, TypeV>
        extends AbstractMap<TypeK, TypeV>
        implements ConcurrentMap<TypeK, TypeV>, Cloneable, Serializable {
    private static final int MIN_SIZE_LOG = 3;
    // 1 << 3 = 8
    private static final int MIN_SIZE = (1 << MIN_SIZE_LOG);

    public NonBlockingHashMap() {
        this(MIN_SIZE);
    }

    public NonBlockingHashMap(final int initial_sz) {
        initialize(initial_sz);
    }
}
public class NonBlockingHashMap<TypeK, TypeV>
        extends AbstractMap<TypeK, TypeV>
        implements ConcurrentMap<TypeK, TypeV>, Cloneable, Serializable {
    private transient long _last_resize_milli;
    private transient Object[] _kvs;

    private final void initialize(int initial_sz) {
        int i;
        // 限制初始大小
        if (initial_sz > 1024 * 1024) initial_sz = 1024 * 1024;
        // i 是 5
        for (i = MIN_SIZE_LOG; (1 << i) < (initial_sz << 2); i++) ;
        // length 66
        _kvs = new Object[((1 << i) << 1) + 2];
        _kvs[0] = new CHM(new ConcurrentAutoTable());
        // length = 32
        _kvs[1] = new int[1 << i];
        _last_resize_milli = System.currentTimeMillis();
    }
}

put 第一个值

put第一个值的过程
  • _kvs[0]: 存储了信息
  • _kvs[1]: 缓存了每个key的 hash 值
public class NonBlockingHashMap<TypeK, TypeV>
        extends AbstractMap<TypeK, TypeV>
        implements ConcurrentMap<TypeK, TypeV>, Cloneable, Serializable {
    public static final Object TOMBSTONE = new Object();
    static volatile int DUMMY_VOLATILE;
    private transient Object[] _kvs;

    @Override
    public TypeV put(TypeK key, TypeV val) {
        return putIfMatch(key, val, NO_MATCH_OLD);
    }

    public final TypeV putIfMatch(Object key, Object newVal, Object oldVal) {
        final Object res = putIfMatch0(this, _kvs, key, newVal, oldVal);
        return res == TOMBSTONE ? null : (TypeV) res;
    }

    private static final Object putIfMatch0(
            final NonBlockingHashMap topmap,
            final Object[] kvs,
            final Object key,
            final Object putval,    // value
            final Object expVal) { // NO_MATCH_OLD
        // key 的 hashCode
        final int fullhash = hash(key);
        // 长度,(length - 2) / 2
        final int len = len(kvs);
        // 控制信息结构体
        final CHM chm = chm(kvs);
        // 缓存的key的hash数组
        final int[] hashes = hashes(kvs);
        // 期望位置
        int idx = fullhash & (len - 1);

        // 重试次数
        int reprobe_cnt = 0;
        Object K = null, V = null;
        Object[] newkvs = null;
        while (true) {
            // 旧的 value
            V = val(kvs, idx);
            // 旧的 key
            K = key(kvs, idx);
            if (K == null) {
                // 说明目标位置没有被占用
                // TOMBSTONE 就是删除的意思,而对应位置已经不存在了,所以直接置空
                if (putval == TOMBSTONE) return TOMBSTONE;
                // 通常用于 replace,而该函数的必要条件是旧值必须存在,然后很明显,旧值不存在。所以操作也失败
                if (expVal == MATCH_ANY) return TOMBSTONE;
                // cas 替换旧值,但是只替换了 key,说白了 value 还没替换
                if (CAS_key(kvs, idx, null, key)) {
                    chm._slots.add(1);
                    // 更新key值对应的hash的数组中
                    hashes[idx] = fullhash;
                    break;
                }
            }
            // ...........
        }
        while (true) {
            // 既然旧值与新值相同,则没必要更新
            if (putval == V) return V;
            if (newkvs == null && // 第一次肯定是 true
                    ((V == null && chm.tableFull(reprobe_cnt, len)) ||
                            V instanceof Prime)) { // 由于 V 是 null,只要 tableFull 不满足,则不会执行 resize
                // 需要resize
                // ...........
            }
            if (newkvs != null) {    // 第一次 newkvs 是 null
                // ...........
            }
            if (expVal != NO_MATCH_OLD &&  // expVal 就是 NO_MATCH_OLD 
                    V != expVal &&
                    (expVal != MATCH_ANY || V == TOMBSTONE || V == null) &&
                    !(V == null && expVal == TOMBSTONE) &&
                    (expVal == null || !expVal.equals(V))) {
                // ............
            }
            // 尝试 cas 设置 v
            if (CAS_val(kvs, idx, V, putval)) break;
            // ...............
        }
        if (expVal != null) { // expVal = NO_MATCH_OLD
            // && 左边成立,且 右边也会成立。因此执行 size + 1
            if ((V == null || V == TOMBSTONE) && putval != TOMBSTONE) chm._size.add(1);
            // .............
        }
        // 返回旧值
        return (V == null && expVal != null) ? TOMBSTONE : V;
    }

}

put 遇到了 key 的 resize

假设容器已经满了,因此无法再放入元素,因此第一步的 key 无法写入成功,因此触发了 resize

key的resize
public class NonBlockingHashMap<TypeK, TypeV>
        extends AbstractMap<TypeK, TypeV>
        implements ConcurrentMap<TypeK, TypeV>, Cloneable, Serializable {
    private static final Object putIfMatch0(
            final NonBlockingHashMap topmap,
            final Object[] kvs,
            final Object key,
            final Object putval,    // value
            final Object expVal) { // NO_MATCH_OLD
        // key 的 hashCode
        final int fullhash = hash(key);
        // 长度,(length - 2) / 2
        final int len = len(kvs);
        // 控制信息结构体
        final CHM chm = chm(kvs);
        // 缓存的key的hash数组
        final int[] hashes = hashes(kvs);
        // 期望位置
        int idx = fullhash & (len - 1);
        // 重试次数
        int reprobe_cnt = 0;
        Object K = null, V = null;
        Object[] newkvs = null;
        while (true) {
            // 旧的 value
            V = val(kvs, idx);
            // 旧的 key
            K = key(kvs, idx);
            if (K == null) {
                // 由于已经满了,所以这里肯定是 K != null
                // ......
            }
            // 先不管 _newkvs 是什么
            newkvs = chm._newkvs;
            // 比较key是否相同,由于已经满了
            if (keyeq(K, key, hashes, idx, fullhash)) {
                break;
            }
            // reprobe_limit(len) 的值是 10,因此 尝试 10 次后,就会触发下面的 resize
            if (++reprobe_cnt >= reprobe_limit(len) || K == TOMBSTONE) {
                // 重新生成一个扩容后的数组
                newkvs = chm.resize(topmap, kvs);
                // newkvs 是个空数组,所以还需要拷贝数据
                if (expVal != null) topmap.help_copy(newkvs);
                return putIfMatch0(topmap, newkvs, key, putval, expVal);
            }
        }
    }
}

reprobe_limit

public class NonBlockingHashMap<TypeK, TypeV>
        extends AbstractMap<TypeK, TypeV>
        implements ConcurrentMap<TypeK, TypeV>, Cloneable, Serializable {
    private static final int REPROBE_LIMIT = 10;

    private static int reprobe_limit(int len) {
        // 右移4次,相当于除以 (2^4 = 16),len = 8 的话,len >> 4 = 0
        return REPROBE_LIMIT + (len >> 4);
    }
}

CHM.resize

public class NonBlockingHashMap<TypeK, TypeV>
        extends AbstractMap<TypeK, TypeV>
        implements ConcurrentMap<TypeK, TypeV>, Cloneable, Serializable {
    private static final class CHM<TypeK, TypeV> {
        volatile Object[] _newkvs;

        private final Object[] resize(NonBlockingHashMap topmap, Object[] kvs) {
            // 这里的 topmap 是当前 map
            // 稍后介绍 _newkvs 赋值,默认是 null
            Object[] newkvs = _newkvs;
            if (newkvs != null) {
                return newkvs;
            }
            // 这里是 8
            int oldlen = len(kvs);
            // sz 是 8
            int sz = size();
            // newsz 是 8
            int newsz = sz;
            if (sz >= (oldlen >> 2)) { // true
                // 经过了复杂计算,最终 newsz 是 32
            }
            // .......... 省略大量计算
            // newkvs 是 null
            newkvs = _newkvs;
            // ...........
            // len 是 32 * 2 + 2 = 66
            newkvs = new Object[(int) len];
            // _size 是 8
            newkvs[0] = new CHM(_size);
            // log2 是 5,最终值是 32
            newkvs[1] = new int[1 << log2];
            // ............
            // cas 后会把 _newkvs 修改为新的数组值
            if (CAS_newkvs(newkvs)) {
                // 什么也不做
                topmap.rehash();
            } else {
            }
            return newkvs;
        }
    }
}

NonBlockingHashMap.help_copy

public class NonBlockingHashMap<TypeK, TypeV>
        extends AbstractMap<TypeK, TypeV>
        implements ConcurrentMap<TypeK, TypeV>, Cloneable, Serializable {
    private transient Object[] _kvs;

    private final Object[] help_copy(Object[] helper) {
        Object[] topkvs = _kvs;
        // old chm
        CHM topchm = chm(topkvs);
        // topkvs 是旧值,上面已经 cas 对旧值赋值了,所以这里不为空
        if (topchm._newkvs == null) return helper;
        topchm.help_copy_impl(this, topkvs, false);
        return helper;
    }
}
public class NonBlockingHashMap<TypeK, TypeV>
        extends AbstractMap<TypeK, TypeV>
        implements ConcurrentMap<TypeK, TypeV>, Cloneable, Serializable {
    private static final class CHM<TypeK, TypeV> {
        volatile long _copyDone = 0;

        private final void help_copy_impl(NonBlockingHashMap topmap, Object[] oldkvs, boolean copy_all) {
            Object[] newkvs = _newkvs;
            // oldlen = 8
            int oldlen = len(oldkvs);
            // MIN_COPY_WORK = 8
            final int MIN_COPY_WORK = Math.min(oldlen, 1024);
            while (_copyDone < oldlen) {
                int workdone = 0;
                for (int i = 0; i < MIN_COPY_WORK; i++) {
                    if (copy_slot(topmap, (copyidx + i) & (oldlen - 1), oldkvs, newkvs)) {
                        workdone++;
                    }
                }

            }
        }
    }
}

小结

  • 采用开放地址法存储数据
  • 如果遇到扩容,则根据算法扩大一定倍数,生成新的数组
  • 还需要把旧数组的数据拷贝到新数组,该机制和 ConcurrentHashMap 一致。
    • 默认情况下:由扩容线程完成拷贝
    • 旧数组长度超过 1024时:扩容线程先拷贝1024个,然后循环拷贝,直到拷贝完成。如果此时有新的线程也执行了 put 操作,则让该线程也参与拷贝工作。

Clone this wiki locally