Spring Data Redis Pipelining 정리

레디스는 파이프라이닝(pipelining)을 지원합니다. 파이프라이닝은 여러 명령을 서버에 응답을 기다리지 않고 한꺼번에 보내고, 나중에 한 번에 응답을 읽어 들이는 방식이다. 이 방식은 예를 들어 같은 리스트에 많은 요소를 추가할 때처럼 여러 명령을 연속으로 보내야 하는 경우 성능을 향상할 수 있다.

 

Spring Data Redis는 파이프라인에서 명령을 실행할 수 있도록 여러 RedisTemplate 메서드를 제공한다. 파이프라인으로 실행되는 작업의 결과를 신경 쓰지 않는다면, execute 메서드를 사용할 수 있으며, 이때 파이프라인 여부를 나타내는 두 번째 인자에 true를 전달하면 된다.

 

반면에, 파이프라인 내에서 실행한 명령의 결과가 필요하다면, executePipelined 메서드를 사용하면 된다. 이 메서드는 전달받은 RedisCallback 또는 SessionCallback을 파이프라인 내에서 실행하고, 실행 결과들을 반환한다. 사용 예제: 

//pop a specified number of items from a queue
List<Object> results = stringRedisTemplate.executePipelined(
  new RedisCallback<Object>() {
    public Object doInRedis(RedisConnection connection) throws DataAccessException {
      StringRedisConnection stringRedisConn = (StringRedisConnection)connection;
      for(int i=0; i< batchSize; i++) {
        stringRedisConn.rPop("myqueue");
      }
    return null;
  }
});

 

주의할 점은, RedisCallback에서 반환하는 값은 반드시 null이어야 한다. 이는 RedisTemplate이 RedisCallback의 반환값을 무시하고, 파이프라인으로 실행된 명령들의 결과를 반환하기 때문이다.

 

// RedisTemplate
public List<Object> executePipelined(RedisCallback<?> action, @Nullable RedisSerializer<?> resultSerializer) {  
    return (List)this.execute((RedisCallback)((connection) -> {  
        connection.openPipeline();  
        boolean pipelinedClosed = false;  

        List var7;  
        try {  
            Object result = action.doInRedis(connection);  
            if (result != null) {  
                throw new InvalidDataAccessApiUsageException("Callback cannot return a non-null value as it gets overwritten by the pipeline");  
            }  

            List<Object> closePipeline = connection.closePipeline();  
            pipelinedClosed = true;  
            var7 = this.deserializeMixedResults(closePipeline, resultSerializer, this.hashKeySerializer, this.hashValueSerializer);  
        } finally {  
            if (!pipelinedClosed) {  
                connection.closePipeline();  
            }  

        }  

        return var7;  
    }));  
}

// LettuceConnection
public List<Object> closePipeline() {  
    if (!this.isPipelined) {  
        return Collections.emptyList();  
    } else {  
        this.pipeliningFlushState.onClose(this.getOrCreateDedicatedConnection());  
        this.pipeliningFlushState = null;  
        this.isPipelined = false;  
        List<CompletableFuture<?>> futures = new ArrayList(this.ppline.size());  

        for(LettuceResult<?, ?> result : this.ppline) {  
            futures.add((CompletableFuture)result.getResultHolder());  
        }  

        try {  
            boolean done = LettuceFutures.awaitAll(this.timeout, TimeUnit.MILLISECONDS, (Future[])futures.toArray(new RedisFuture[0]));  
            List<Object> results = new ArrayList(futures.size());  
            Exception problem = null;  
            if (done) {  
                for(LettuceResult<?, ?> result : this.ppline) {  
                    CompletableFuture<?> resultHolder = (CompletableFuture)result.getResultHolder();  
                    if (resultHolder.isCompletedExceptionally()) {  
                        // ...
                        }  

                        Exception exception = new InvalidDataAccessApiUsageException(message);  
                        if (problem == null) {  
                            problem = exception;  
                        }  
                        results.add(exception);  
                    } else if (!result.isStatus()) {  
                        // result가 정상적인 경우
                        try {  
                            results.add(result.conversionRequired() ? result.convert(result.get()) : result.get()); 
                        } catch (DataAccessException var13) {  
                            if (problem == null) {  
                                problem = var13;  
                            }  

                            results.add(var13);  
                        }  
                    }  
                }  
            }  
            // ..
        } catch (Exception ex) {  
            throw new RedisPipelineException(ex);  
        }  
    }  
}

이를 자세히 알아보기 위해 RedisTemplate의 executePipelined 메서드 구현을 살펴보면 result가 null인지 검증하는 것을 확인할 수 있다. 추가로 connection.closePipeline()를 통해 파이프라인을 닫으면서 RedisCallback에서 사용한 명령에 대한 응답을 반환하고 있는 것을 확인할 수 있다.

이러한 이유로 RedisCallback에서 null을 응답 함에도 executePipelined에서는 원하는 값을 응답받을 수 있다.

 

•  참고자료
  • 공식문서: https://docs.spring.io/spring-data/redis/reference/redis/pipelining.html