.

reader, processor, writer 100 .

read .

1. Map : Map .

2. Redis Insert: Redis .

parallelStream

parallelStream .

parallelStream stream . .

items.forEach(item -> ( do something))items.parallelStream().forEach(item -> ( do something))

parallelStream

parallelStream

1. : parallelStream , . .

2. : parallelStream . .

3. ForkJoinPool : parallelStream ForkJoinPool.commonPool() . JVM CPU , .

parallelStream , , ...

~

ForkJoinPool

ForkJoinPool parallelStream .

.

private final ForkJoinPool forkJoinPool = new ForkJoinPool(6)try (  forkJoinPool.submit(() ->    keyInItems.parallelStream().forEach(item ->      keyInItemMap.putIfAbsent(item.getQuery(), item)    )  ).get()) catch (InterruptedException | ExecutionException e) (  log.error("Error during parallel processing", e)  throw new RuntimeException("Error during parallel processing", e))

Redis - executePipelined

Redis , parallelStream executePipelined .

executePipelined .

db insert batch insert .

try (  redisTemplate.executePipelined((RedisCallback<Object>) connection -> (    items.forEach(item -> (      try (        String key = REDIS_TYPE_NAME + item.getQuery()        byte[] value = objectMapper.writeValueAsBytes(item)        connection.set(key.getBytes(), value)        connection.expire(key.getBytes(), TimeUnit.DAYS.toSeconds(2))      ) catch (JsonProcessingException e) (        log.error("()", e.getMessage())        throw new RuntimeException("Redis Pipeline   ", e)      )    ))    return null  ))) catch (Exception e) (  log.error("Error during Redis Pipeline operation", e)  throw new RuntimeException("Redis Pipeline   ", e))

: 6~7

: 30

parallelStream Redis executePipelined , .

.

( parallelStream) , .

, ..!

ParallelStream을 활용한 병렬 처리로 배치 성능 개선하기

댓글 0개