RedisTemplate 使用 pipeline 时需要注意的问题

RedisTemplate 使用 pipeline 时需要注意的问题

RedisTemplate 使用 pipeline 进行批量 set 时，需要把 key 和 value 都转为字节

1. 直接使用 getBytes() 转为字节，在读取数据时，会抛出以下序列化异常

//错误代码
protected void process(ReconRedisContext reconRedisContext) {
        String key = this.getLockKey(reconRedisContext);
        Pagination pagination = this.queryReconData(reconRedisContext);
        if (ObjectUtils.isNotEmpty(pagination)
                && CollectionUtils.isNotEmpty(pagination.getList())) {
            List<?> resList = pagination.getList();
            redisTemplate.executePipelined(new RedisCallback<Set<?>>() {
                @Override
                public Set<?> doInRedis(RedisConnection connection) throws DataAccessException {
                    resList.stream().forEach(value -> {
                         connection.sAdd(key.getBytes(StandardCharsets.UTF_8), value.toString().getBytes(StandardCharsets.UTF_8));
                    });
                    return null;
                }
            });
        }
    }

// 抛出序列化异常
org.springframework.data.redis.serializer.SerializationException: Could not read JSON: Unrecognized token 'value': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

2. 需要使用 RedisTemplate 已经设置的 Serializer ，将key 和 value 序列化成byte数据，代码如下：

protected void process(ReconRedisContext reconRedisContext) {
        RedisSerializer keySerializer = redisTemplate.getKeySerializer();
        RedisSerializer valueSerializer = redisTemplate.getValueSerializer();

        String key = this.getLockKey(reconRedisContext);
        Pagination pagination = this.queryReconData(reconRedisContext);
        if (ObjectUtils.isNotEmpty(pagination)
                && CollectionUtils.isNotEmpty(pagination.getList())) {
            List<?> resList = pagination.getList();
            redisTemplate.executePipelined(new RedisCallback<Set<?>>() {
                @Override
                public Set<?> doInRedis(RedisConnection connection) throws DataAccessException {
                    resList.stream().forEach(value -> {
                        connection.sAdd(keySerializer.serialize(key), valueSerializer.serialize(value));
                    });
                    return null;
                }
            });
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

3. 异常原因：

// RedisTemplate 默认使用 DefaultSetOperations 存放数据的源码如下：
public Long add(K key, V... values) {

		byte[] rawKey = rawKey(key);
		byte[][] rawValues = rawValues((Object[]) values);
		return execute(connection -> connection.sAdd(rawKey, rawValues));
}

byte[] rawKey(Object key) {

		Assert.notNull(key, "non null key required");

		if (keySerializer() == null && key instanceof byte[]) {
			return (byte[]) key;
		}

		return keySerializer().serialize(key);
}

byte[] rawValue(Object value) {

		if (valueSerializer() == null && value instanceof byte[]) {
			return (byte[]) value;
		}

		return valueSerializer().serialize(value);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

看源码，RedisTemplate 也是把 key 和 value 都转为了字节，但是使用了我们自己设置的 Serializer ，所以，我们在使用 pipeline 时，也需要使用我们设置的 Serializer。

注意：根据源码来看，Redis 的其他数据结构，使用 pipeline 时，也会存在序列化的问题，在代码编写的时候，需要注意。
源自