Enhance RedisItemReader/Writer performance with pipelining

Signed-off-by: Hyunwoo Jung <hyunwoojung@kakao.com>

Author: noojung

spring-batch-infrastructure/src/main/java/org/springframework/batch/infrastructure/item/redis/RedisItemReader.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2023-2025 the original author or authors.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -20,21 +20,31 @@
 import org.springframework.batch.infrastructure.item.ExecutionContext;
 import org.springframework.batch.infrastructure.item.ItemStreamException;
 import org.springframework.batch.infrastructure.item.ItemStreamReader;
+import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.data.redis.core.SessionCallback;
 import org.springframework.util.Assert;
 
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+
 /**
  * Item reader for Redis based on Spring Data Redis. Uses a {@link RedisTemplate} to query
  * data. The user should provide a {@link ScanOptions} to specify the set of keys to
- * query.
+ * query. The {@code fetchSize} property controls how many items are fetched from Redis in
+ * a single pipeline round-trip for efficiency.
  *
  * <p>
  * The implementation is not thread-safe and not restartable.
  * </p>
  *
  * @author Mahmoud Ben Hassine
+ * @author Hyunwoo Jung
  * @since 5.1
  * @param <K> type of keys
  * @param <V> type of values
@@ -47,28 +57,32 @@ public class RedisItemReader<K, V> implements ItemStreamReader<V> {
 
 	private @Nullable Cursor<K> cursor;
 
-	public RedisItemReader(RedisTemplate<K, V> redisTemplate, ScanOptions scanOptions) {
+	private final int fetchSize;
+
+	private final Deque<V> buffer;
+
+	public RedisItemReader(RedisTemplate<K, V> redisTemplate, ScanOptions scanOptions, int fetchSize) {
 		Assert.notNull(redisTemplate, "redisTemplate must not be null");
 		Assert.notNull(scanOptions, "scanOptions must no be null");
+		Assert.isTrue(fetchSize > 0, "fetchSize must be greater than 0");
 		this.redisTemplate = redisTemplate;
 		this.scanOptions = scanOptions;
+		this.fetchSize = fetchSize;
+		this.buffer = new ArrayDeque<>();
 	}
 
 	@Override
 	public void open(ExecutionContext executionContext) throws ItemStreamException {
 		this.cursor = this.redisTemplate.scan(this.scanOptions);
 	}
 
-	@SuppressWarnings("DataFlowIssue")
 	@Override
 	public @Nullable V read() throws Exception {
-		if (this.cursor.hasNext()) {
-			K nextKey = this.cursor.next();
-			return this.redisTemplate.opsForValue().get(nextKey);
-		}
-		else {
-			return null;
+		if (this.buffer.isEmpty()) {
+			fetchNext();
 		}
+
+		return this.buffer.pollFirst();
 	}
 
 	@SuppressWarnings("DataFlowIssue")
@@ -77,4 +91,36 @@ public void close() throws ItemStreamException {
 		this.cursor.close();
 	}
 
+	@SuppressWarnings("DataFlowIssue")
+	private void fetchNext() {
+		List<K> keys = new ArrayList<>();
+		while (this.cursor.hasNext() && keys.size() < this.fetchSize) {
+			keys.add(this.cursor.next());
+		}
+
+		if (keys.isEmpty()) {
+			return;
+		}
+
+		@SuppressWarnings("unchecked")
+		List<V> items = (List<V>) this.redisTemplate.executePipelined(sessionCallback(keys));
+
+		this.buffer.addAll(items);
+	}
+
+	private SessionCallback<Object> sessionCallback(List<K> keys) {
+		return new SessionCallback<>() {
+
+			@SuppressWarnings("NullAway")
+			@Override
+			public @Nullable Object execute(RedisOperations operations) throws DataAccessException {
+				for (K key : keys) {
+					operations.opsForValue().get(key);
+				}
+
+				return null;
+			}
+		};
+	}
+
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/infrastructure/item/redis/RedisItemWriter.java

@@ -16,12 +16,20 @@
 
 package org.springframework.batch.infrastructure.item.redis;
 
+import org.jspecify.annotations.Nullable;
 import org.springframework.batch.infrastructure.item.ItemWriter;
 import org.springframework.batch.infrastructure.item.KeyValueItemWriter;
 import org.springframework.core.convert.converter.Converter;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.SessionCallback;
+import org.springframework.data.util.Pair;
 import org.springframework.util.Assert;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * <p>
  * An {@link ItemWriter} implementation for Redis using a {@link RedisTemplate} .
@@ -30,12 +38,15 @@
  * @author Santiago Molano
  * @author Mahmoud Ben Hassine
  * @author Stefano Cordio
+ * @author Hyunwoo Jung
  * @since 5.1
  */
 public class RedisItemWriter<K, T> extends KeyValueItemWriter<K, T> {
 
 	private RedisTemplate<K, T> redisTemplate;
 
+	private final List<Pair<K, T>> buffer = new ArrayList<>();
+
 	/**
 	 * Create a new {@link RedisItemWriter}.
 	 * @param itemKeyMapper the {@link Converter} used to derive a key from an item.
@@ -50,19 +61,25 @@ public RedisItemWriter(Converter<T, K> itemKeyMapper, RedisTemplate<K, T> redisT
 
 	@Override
 	protected void writeKeyValue(K key, T value) {
-		if (this.delete) {
-			this.redisTemplate.delete(key);
-		}
-		else {
-			this.redisTemplate.opsForValue().set(key, value);
-		}
+		this.buffer.add(Pair.of(key, value));
 	}
 
 	@Override
 	protected void init() {
 		Assert.notNull(this.redisTemplate, "RedisTemplate must not be null");
 	}
 
+	@Override
+	protected void flush() throws Exception {
+		if (this.buffer.isEmpty()) {
+			return;
+		}
+
+		this.redisTemplate.executePipelined(sessionCallback());
+
+		this.buffer.clear();
+	}
+
 	/**
 	 * Set the {@link RedisTemplate} to use.
 	 * @param redisTemplate the template to use
@@ -71,4 +88,33 @@ public void setRedisTemplate(RedisTemplate<K, T> redisTemplate) {
 		this.redisTemplate = redisTemplate;
 	}
 
+	private SessionCallback<Object> sessionCallback() {
+		return new SessionCallback<>() {
+
+			@SuppressWarnings({ "unchecked", "NullAway" })
+			@Override
+			public @Nullable Object execute(RedisOperations operations) throws DataAccessException {
+				if (RedisItemWriter.this.delete) {
+					executeDeleteOperations(operations);
+				}
+				else {
+					executeSetOperations(operations);
+				}
+				return null;
+			}
+		};
+	}
+
+	private void executeDeleteOperations(RedisOperations<K, T> operations) {
+		for (Pair<K, T> item : this.buffer) {
+			operations.delete(item.getFirst());
+		}
+	}
+
+	private void executeSetOperations(RedisOperations<K, T> operations) {
+		for (Pair<K, T> item : this.buffer) {
+			operations.opsForValue().set(item.getFirst(), item.getSecond());
+		}
+	}
+
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/infrastructure/item/redis/builder/RedisItemReaderBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2023-2025 the original author or authors.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@
  * Builder for {@link RedisItemReader}.
  *
  * @author Mahmoud Ben Hassine
+ * @author Hyunwoo Jung
  * @since 5.1
  * @param <K> type of keys
  * @param <V> type of values
@@ -33,6 +34,8 @@ public class RedisItemReaderBuilder<K, V> {
 
 	private ScanOptions scanOptions;
 
+	private int fetchSize;
+
 	/**
 	 * Set the {@link RedisTemplate} to use in the reader.
 	 * @param redisTemplate the template to use
@@ -53,12 +56,22 @@ public RedisItemReaderBuilder<K, V> scanOptions(ScanOptions scanOptions) {
 		return this;
 	}
 
+	/**
+	 * Set the fetchSize to how many items from Redis in a single round-trip.
+	 * @param fetchSize the number of items to fetch per pipeline execution
+	 * @return the current builder instance for fluent chaining
+	 */
+	public RedisItemReaderBuilder<K, V> fetchSize(int fetchSize) {
+		this.fetchSize = fetchSize;
+		return this;
+	}
+
 	/**
 	 * Build a new {@link RedisItemReader}.
 	 * @return a new item reader
 	 */
 	public RedisItemReader<K, V> build() {
-		return new RedisItemReader<>(this.redisTemplate, this.scanOptions);
+		return new RedisItemReader<>(this.redisTemplate, this.scanOptions, this.fetchSize);
 	}
 
 }

spring-batch-infrastructure/src/test/java/org/springframework/batch/infrastructure/item/redis/RedisItemReaderIntegrationTests.java

@@ -52,7 +52,7 @@
 @ExtendWith(SpringExtension.class)
 class RedisItemReaderIntegrationTests {
 
-	private static final DockerImageName REDIS_IMAGE = DockerImageName.parse("redis:8.0.3");
+	private static final DockerImageName REDIS_IMAGE = DockerImageName.parse("redis:8.2.2");
 
 	@Container
 	public static RedisContainer redis = new RedisContainer(REDIS_IMAGE);
@@ -82,7 +82,7 @@ void testRead(RedisConnectionFactory connectionFactory) throws Exception {
 
 		RedisTemplate<String, Person> redisTemplate = setUpRedisTemplate(connectionFactory);
 		ScanOptions scanOptions = ScanOptions.scanOptions().match("person:*").count(10).build();
-		this.reader = new RedisItemReader<>(redisTemplate, scanOptions);
+		this.reader = new RedisItemReader<>(redisTemplate, scanOptions, 10);
 
 		this.reader.open(new ExecutionContext());
 

spring-batch-infrastructure/src/test/java/org/springframework/batch/infrastructure/item/redis/RedisItemReaderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2023-2025 the original author or authors.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -16,21 +16,26 @@
 package org.springframework.batch.infrastructure.item.redis;
 
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Answers;
 import org.mockito.Mock;
-import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;
-
 import org.springframework.batch.infrastructure.item.ExecutionContext;
-import org.springframework.batch.infrastructure.item.redis.RedisItemReader;
 import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.data.redis.core.SessionCallback;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
-public class RedisItemReaderTests {
+class RedisItemReaderTests {
 
 	@Mock(answer = Answers.RETURNS_DEEP_STUBS)
 	private RedisTemplate<String, String> redisTemplate;
@@ -41,15 +46,35 @@ public class RedisItemReaderTests {
 	@Mock
 	private Cursor<String> cursor;
 
+	private List<String> results;
+
+	@BeforeEach
+	void setUp() {
+		this.results = new ArrayList<>();
+		when(this.redisTemplate.executePipelined(any(SessionCallback.class))).thenAnswer(invocation -> {
+			SessionCallback<?> sessionCallback = invocation.getArgument(0);
+			sessionCallback.execute(this.redisTemplate);
+			return this.results;
+		});
+	}
+
 	@Test
 	void testRead() throws Exception {
 		// given
-		Mockito.when(this.redisTemplate.scan(this.scanOptions)).thenReturn(this.cursor);
-		Mockito.when(this.cursor.hasNext()).thenReturn(true, true, false);
-		Mockito.when(this.cursor.next()).thenReturn("person:1", "person:2");
-		Mockito.when(this.redisTemplate.opsForValue().get("person:1")).thenReturn("foo");
-		Mockito.when(this.redisTemplate.opsForValue().get("person:2")).thenReturn("bar");
-		RedisItemReader<String, String> redisItemReader = new RedisItemReader<>(this.redisTemplate, this.scanOptions);
+		when(this.redisTemplate.scan(this.scanOptions)).thenReturn(this.cursor);
+		when(this.cursor.hasNext()).thenReturn(true, true, false);
+		when(this.cursor.next()).thenReturn("person:1", "person:2");
+		when(this.redisTemplate.opsForValue().get("person:1")).thenAnswer(invocation -> {
+			results.add("foo");
+			return null;
+		});
+		when(this.redisTemplate.opsForValue().get("person:2")).thenAnswer(invocation -> {
+			results.add("bar");
+			return null;
+		});
+
+		RedisItemReader<String, String> redisItemReader = new RedisItemReader<>(this.redisTemplate, this.scanOptions,
+				10);
 		redisItemReader.open(new ExecutionContext());
 
 		// when

spring-batch-infrastructure/src/test/java/org/springframework/batch/infrastructure/item/redis/RedisItemWriterIntegrationTests.java

@@ -48,7 +48,7 @@
 @ExtendWith(SpringExtension.class)
 class RedisItemWriterIntegrationTests {
 
-	private static final DockerImageName REDIS_IMAGE = DockerImageName.parse("redis:8.0.3");
+	private static final DockerImageName REDIS_IMAGE = DockerImageName.parse("redis:8.2.2");
 
 	@Container
 	public static RedisContainer redis = new RedisContainer(REDIS_IMAGE);