聊聊redisson的RMap的computeIfAbsent操作

序

本文主要研究一下redisson的RMap的computeIfAbsent操作

实例

@Test
    public void testRMapComputeIfAbsent(){
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://192.168.99.100:6379");
        RedissonClient redisson = Redisson.create(config);
        RMap<String, CityInfo> map = redisson.getMap("anyMap");

        CityInfo bj = CityInfo.builder().name("bj").build();
        CityInfo currentObject = map.computeIfAbsent("bj", k -> bj);
        Assert.assertEquals(bj.toString(),currentObject.toString());
    }

源码分析

ConcurrentMap.computeIfAbsent

java/util/concurrent/ConcurrentMap.java

/**
     * {@inheritDoc}
     *
     * @implSpec
     * The default implementation is equivalent to the following steps for this
     * {@code map}, then returning the current value or {@code null} if now
     * absent:
     *
     * <pre> {@code
     * if (map.get(key) == null) {
     *     V newValue = mappingFunction.apply(key);
     *     if (newValue != null)
     *         return map.putIfAbsent(key, newValue);
     * }
     * }</pre>
     *
     * The default implementation may retry these steps when multiple
     * threads attempt updates including potentially calling the mapping
     * function multiple times.
     *
     * <p>This implementation assumes that the ConcurrentMap cannot contain null
     * values and {@code get()} returning null unambiguously means the key is
     * absent. Implementations which support null values <strong>must</strong>
     * override this default implementation.
     *
     * @throws UnsupportedOperationException {@inheritDoc}
     * @throws ClassCastException {@inheritDoc}
     * @throws NullPointerException {@inheritDoc}
     * @since 1.8
     */
    @Override
    default V computeIfAbsent(K key,
            Function<? super K, ? extends V> mappingFunction) {
        Objects.requireNonNull(mappingFunction);
        V v, newValue;
        return ((v = get(key)) == null &&
                (newValue = mappingFunction.apply(key)) != null &&
                (v = putIfAbsent(key, newValue)) == null) ? newValue : v;
    }

• computeIfAbsent当该key不存在时，返回的是新值，而非null
• computeIfAbsent方法里头调用了putIfAbsent

RedissonMap.putIfAbsent

redisson-3.8.1-sources.jar!/org/redisson/RedissonMap.java

@Override
    public V putIfAbsent(K key, V value) {
        return get(putIfAbsentAsync(key, value));
    }

    @Override
    public RFuture<V> putIfAbsentAsync(final K key, final V value) {
        checkKey(key);
        checkValue(key);
        
        RFuture<V> future = putIfAbsentOperationAsync(key, value);
        if (hasNoWriter()) {
            return future;
        }
        
        MapWriterTask<V> listener = new MapWriterTask<V>() {
            @Override
            public void execute() {
                options.getWriter().write(key, value);
            }
            
            @Override
            protected boolean condition(Future<V> future) {
                return future.getNow() == null;
            }

        };
        return mapWriterFuture(future, listener);
    }

    protected boolean hasNoWriter() {
        return options == null || options.getWriter() == null;
    }

    protected RFuture<V> putIfAbsentOperationAsync(K key, V value) {
        return commandExecutor.evalWriteAsync(getName(key), codec, RedisCommands.EVAL_MAP_VALUE,
                 "if redis.call('hsetnx', KEYS[1], ARGV[1], ARGV[2]) == 1 then "
                    + "return nil "
                + "else "
                    + "return redis.call('hget', KEYS[1], ARGV[1]) "
                + "end",
                Collections.<Object>singletonList(getName(key)), encodeMapKey(key), encodeMapValue(value));
    }

    protected <M> RFuture<M> mapWriterFuture(RFuture<M> future, final MapWriterTask<M> listener) {
        if (options != null && options.getWriteMode() == WriteMode.WRITE_BEHIND) {
            future.addListener(new MapWriteBehindListener<M>(commandExecutor, listener, writeBehindCurrentThreads, writeBehindTasks, options.getWriteBehindThreads()));
            return future;
        }        

        final RPromise<M> promise = new RedissonPromise<M>();
        future.addListener(new FutureListener<M>() {
            @Override
            public void operationComplete(final Future<M> future) throws Exception {
                if (!future.isSuccess()) {
                    promise.tryFailure(future.cause());
                    return;
                }

                if (listener.condition(future)) {
                    commandExecutor.getConnectionManager().getExecutor().execute(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                listener.execute();
                            } catch (Exception e) {
                                promise.tryFailure(e);
                                return;
                            }
                            promise.trySuccess(future.getNow());
                        }
                    });
                } else {
                    promise.trySuccess(future.getNow());
                }
            }
        });

        return promise;
    }

• RedissonMap覆盖了putIfAbsent方法，这里调用的是putIfAbsentAsync，该方法主要调用putIfAbsentOperationAsync
• putIfAbsentOperationAsync使用了一段lua脚本来保证原子性，如果hsetnx之前的key不存在且设置成功则返回nil，否则查找已有的值返回
• putIfAbsentAsync除了调用putIfAbsentOperationAsync，还增加了writer的逻辑，如果有设置writer，则会在putIfAbsentOperationAsync的future成功时回调触发writer
• writer主要用于外部存储用，比如旁路存储一份到本地磁盘，有同步操作及异步操作两种模式

小结

redisson对redis的数据结构操作进行了更进一步的封装，比如redisson的RMap实现了java.util.concurrent.ConcurrentMap接口和java.util.Map接口，实现了诸如putIfAbsent的方法，用lua脚本在服务端保证了操作的原子性。

doc

• 7. 分布式集合
• 阿里云专访Redisson作者Rui Gu：构建开源企业级Redis客户端之路