Redis管道机制（pipeline）

Redis管道机制（pipeline）

 

    Redis的管道机制，其实是为了批量读写而设计的，如果进行多次的读和写数据到redis，每次都建立一个链接，这样是比较消耗资源的，而且也比较忙，于是想到了管道机制（pipeline），只建立一个连接，然后批量执行读或写。

 

 

 

 

插入数据效果

 

      非管道： 数据量 = 2W， 执行时间 = 1分多钟

      管道：     数据量 = 2W， 执行时间 = 1~2秒

 

 

 

 

代码讲解

 

 

    public boolean add(final List<T> list) {
        final boolean result = stringRedisTemplate.execute(new RedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(final RedisConnection connection)
                    throws DataAccessException {
                for (final T object : list) {
                    final Map<String, String> hash = RedisMapUtil.toMap(object);
                    final Map<byte[], byte[]> hashes = new LinkedHashMap<byte[], byte[]>(hash.size());
                    for (final Map.Entry<String, String> entry : hash.entrySet()) {
                        hashes.put(rawHashKey(entry.getKey()), rawHashValue(entry.getValue()));
                    }
                    final byte[] key = stringRedisTemplate.getStringSerializer().serialize(getDefaultKey(object.getId()));
                    connection.hMSet(key, hashes);
                }
                return true;
            }
        }, false, true);
        return result;
    }

    public boolean addString(final Map<String, String> map) {
        final boolean result = stringRedisTemplate.execute(new RedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(final RedisConnection connection)
                    throws DataAccessException {
                for (final Map.Entry<String, String> entry : map.entrySet()) {
                    final byte[] key = stringRedisTemplate.getStringSerializer().serialize(entry.getKey());
                    final byte[] value = stringRedisTemplate.getStringSerializer().serialize(entry.getValue());
                    connection.set(key, value);
                }
                return true;
            }
        }, false, true);
        return result;
    }

    @SuppressWarnings("unchecked")
    private <HK> byte[] rawHashKey(final HK hashKey) {
        Assert.notNull(hashKey, "non null hash key required");
        if (stringRedisTemplate.getHashKeySerializer() == null && hashKey instanceof byte[]) {
            return (byte[]) hashKey;
        }

        final RedisSerializer<HK> serializer = (RedisSerializer<HK>) stringRedisTemplate.getHashKeySerializer();
        return serializer.serialize(hashKey);
    }

    @SuppressWarnings("unchecked")
    private <HV> byte[] rawHashValue(final HV value) {
        if (stringRedisTemplate.getHashValueSerializer() == null & value instanceof byte[]) {
            return (byte[]) value;
        }

        final RedisSerializer<HV> serializer = (RedisSerializer<HV>) stringRedisTemplate.getHashValueSerializer();
        return serializer.serialize(value);
    }

 

 

上面是自己写的方法

• 首先是调用redisTemplate.execute(RedisCallback<T> action, boolean exposeConnection, boolean pipeline)，这里注意的是 exposeConnection = false，pipeline = true
• 然后，第一参数中new了一个内部类，里面的方法接收一个connection，就是我们要批量执行的一个连接
• 如果要保存的是对象，还是要先将对象转成Map<String, String>，然后在转成byte[]，而key是String转byte[]就简单好多
• 最后，如果是对象就调用connection.hMSet(key, hashes);，如果不是就调用connection.set(key, value);
• 有一个很特别的地方就是，一直往connection中set内容，最后才一次过执行。

 

 

 

参考

http://xinklabi.iteye.com/blog/2195547