Boost Performance in Spring Boot with a Single Batch‑Processing Annotation

Problem Statement

In distributed systems and high‑concurrency scenarios, processing each request individually can exhaust database connections, increase network I/O, and waste CPU cycles.

Solution Overview

A batch processing mechanism is built using a custom annotation @BatchProcess combined with Spring AOP. The annotation defines three attributes: threshold – batch size (default 100) timeout – maximum wait time in milliseconds (default 500) processor – the DataProcessor implementation that handles the batch

Custom Annotation

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface BatchProcess {
    int threshold() default 100;
    long timeout() default 500;
    Class<? extends DataProcessor> processor() default DataProcessor.class;
}

Metadata Extraction

During container startup a bean post‑processor scans all beans for methods annotated with @BatchProcess. For each method it extracts the annotation values and stores a BatchMetadata instance in a static map and in Redis for durability.

public class BatchMetadata<T> {
    private String id;
    private String key;
    private int threshold;
    private long timeout;
    private String beanClass;
    private T data;
    // getters, setters, constructor omitted
}

@Component
public class BatchBeanPostProcessor implements InstantiationAwareBeanPostProcessor {
    public static final Map<String, BatchMetadata<Object>> batchs = new ConcurrentHashMap<>();
    @Override
    public PropertyValues postProcessProperties(PropertyValues pvs, Object bean, String beanName) {
        Class<?> targetClass = bean.getClass();
        do {
            ReflectionUtils.doWithLocalMethods(targetClass, method -> {
                MergedAnnotation<?> ann = findValueAnnotation(method);
                if (ann != null && !Modifier.isStatic(method.getModifiers())) {
                    String key = ann.getString("key");
                    int threshold = ann.getInt("threshold");
                    long timeout = ann.getLong("timeout");
                    Class<?> beanClass = ann.getClass("processor");
                    batchs.put(key, new BatchMetadata<>(key, threshold, timeout, beanClass.getName(), null));
                }
            });
            targetClass = targetClass.getSuperclass();
        } while (targetClass != null && targetClass != Object.class);
        return pvs;
    }
    private MergedAnnotation<?> findValueAnnotation(AccessibleObject ao) {
        MergedAnnotations annotations = MergedAnnotations.from(ao);
        MergedAnnotation<?> annotation = annotations.get(BatchProcess.class);
        return annotation.isPresent() ? annotation : null;
    }
}

Processor Interface

/** Generic batch data processor */
public interface DataProcessor<BatchRequest, R> {
    /** Process a list of batch requests */
    R process(List<BatchRequest> datas);
}

Request Wrapper

public class BatchRequest<T, R> {
    private final T data;
    private final String id; // unique identifier
    private final CompletableFuture<R> future;
    public BatchRequest(T data, String id) {
        this.data = data;
        this.id = id;
        this.future = new CompletableFuture<>();
    }
    // getters omitted
}

Example Processor

@Component
public class LogDataProcessor implements DataProcessor<BatchRequest<Log, String>, Void> {
    private static final Logger logger = LoggerFactory.getLogger(LogDataProcessor.class);
    @Override
    public Void process(List<BatchRequest<Log, String>> datas) {
        logger.info("Batch saving: {} log entries", datas.size());
        for (BatchRequest<Log, String> request : datas) {
            request.getFuture().complete(request.getData().source() + " - success");
        }
        return null;
    }
}

Batch Aspect

@Aspect
@Component
public class BatchProcessAspect implements ApplicationContextAware, SmartInitializingSingleton {
    private ApplicationContext context;
    private final Map<String, BlockingQueue<BatchRequest<Object, Object>>> queues = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler;
    private final StringRedisTemplate stringRedisTemplate;
    private final ObjectMapper objectMapper;

    public BatchProcessAspect(StringRedisTemplate stringRedisTemplate, ObjectMapper objectMapper) {
        this.scheduler = Executors.newScheduledThreadPool(BatchBeanPostProcessor.batchs.size());
        this.stringRedisTemplate = stringRedisTemplate;
        this.objectMapper = objectMapper;
    }

    @Around("@annotation(batchProcess)")
    public Object processBatchRequest(ProceedingJoinPoint pjp, BatchProcess batchProcess) throws Throwable {
        // No arguments – invoke directly
        if (pjp.getArgs().length == 0) return pjp.proceed();
        // No custom processor – invoke directly
        if (batchProcess.processor() == DataProcessor.class) return pjp.proceed();

        String key = batchProcess.key();
        BatchMetadata<Object> origin = BatchBeanPostProcessor.batchs.get(key);
        BatchMetadata<Object> metadata = new BatchMetadata<>(origin.getKey(), origin.getThreshold(), origin.getTimeout(), origin.getBeanClass(), origin.getData());

        BlockingQueue<BatchRequest<Object, Object>> queue = queues.computeIfAbsent(key, k -> {
            BlockingQueue<BatchRequest<Object, Object>> q = new LinkedBlockingQueue<>();
            scheduler.scheduleAtFixedRate(() -> processQueueIfNeeded(k, q, metadata), 0, batchProcess.timeout(), TimeUnit.MILLISECONDS);
            return q;
        });

        Object arg = pjp.getArgs()[0];
        metadata.setData(arg);
        metadata.setId(UUID.randomUUID().toString().replace("-", ""));
        BatchRequest<Object, Object> request = new BatchRequest<>(arg, metadata.getId());

        // Persist metadata to Redis (may become a bottleneck under heavy load)
        stringRedisTemplate.opsForHash().put(key, metadata.getId(), objectMapper.writeValueAsString(metadata));
        queue.put(request);

        // Trigger processing when threshold reached
        if (queue.size() >= batchProcess.threshold()) {
            processQueue(queue, metadata);
        }
        // Wait for batch result (blocks the caller)
        return request.getFuture().get();
    }

    private void processQueueIfNeeded(String methodKey, BlockingQueue<BatchRequest<Object, Object>> queue, BatchMetadata<Object> metadata) {
        if (!queue.isEmpty()) {
            processQueue(queue, metadata);
        }
    }

    private void processQueue(BlockingQueue<BatchRequest<Object, Object>> queue, BatchMetadata<Object> metadata) {
        List<BatchRequest<Object, Object>> requests = new ArrayList<>();
        queue.drainTo(requests, metadata.getThreshold());
        if (requests.isEmpty()) return;

        List<Object> dataList = new ArrayList<>(requests.size());
        for (BatchRequest<Object, Object> request : requests) {
            dataList.add(request);
        }
        try {
            Class<?> beanClass = ClassUtils.forName(metadata.getBeanClass(), this.getClass().getClassLoader());
            @SuppressWarnings("unchecked")
            DataProcessor<Object, Object> processor = (DataProcessor<Object, Object>) context.getBean(beanClass);
            processor.process(dataList);
        } catch (Exception e) {
            // exception handling omitted for brevity
        }
        // Clean up Redis entries for successfully processed requests
        List<Object> hashKeys = new ArrayList<>();
        for (BatchRequest<Object, Object> request : requests) {
            if (!request.getFuture().isCompletedExceptionally()) {
                hashKeys.add(request.getId());
            }
        }
        if (!hashKeys.isEmpty()) {
            stringRedisTemplate.opsForHash().delete(metadata.getKey(), hashKeys.toArray());
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext context) {
        this.context = context;
    }

    @Override
    public void afterSingletonsInstantiated() {
        BatchBeanPostProcessor.batchs.keySet().forEach(key -> {
            Map<Object, Object> entries = stringRedisTemplate.opsForHash().entries(key);
            entries.values().forEach(value -> {
                try {
                    BatchMetadata<Object> metadata = objectMapper.readValue(value.toString(), BatchMetadata.class);
                    BatchRequest<Object, Object> request = new BatchRequest<>(metadata.getData(), metadata.getId());
                    BlockingQueue<BatchRequest<Object, Object>> queue = queues.computeIfAbsent(key, k -> {
                        BlockingQueue<BatchRequest<Object, Object>> q = new LinkedBlockingQueue<>();
                        scheduler.scheduleAtFixedRate(() -> processQueueIfNeeded(k, q, metadata), 0, metadata.getTimeout(), TimeUnit.MILLISECONDS);
                        return q;
                    });
                    queue.put(request);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        });
    }
}

Business Method Example

@Service
public class LogService {
    @BatchProcess(threshold = 5, timeout = 2000, processor = LogDataProcessor.class)
    public String log(Log log) {
        return null;
    }
}

Test Case

@Resource
private LogService logService;

@Test
public void testLog() throws Exception {
    final int MAX = 14;
    Thread[] t = new Thread[MAX];
    for (int i = 0; i < MAX; i++) {
        final int idx = i;
        t[i] = new Thread(() -> {
            String ret = logService.log(new Log("c - " + idx, "s - " + idx, LocalDateTime.now()));
            System.err.println(ret);
        });
    }
    for (Thread thread : t) thread.start();
    for (Thread thread : t) thread.join();
}

The test creates 14 concurrent log requests. With threshold=5 the requests are grouped into three batches (5, 5, 4). Each batch is persisted to Redis, processed by LogDataProcessor, and the corresponding Redis entries are removed. The console output shows the completion messages returned by the futures.

Key Mechanisms

Batch queue : Requests are enqueued until the configured threshold is reached.

Timed fallback : A scheduled task runs at the configured timeout interval; it processes whatever is in the queue even if the threshold is not met.

Persistence : Metadata is stored in Redis before processing to avoid data loss.

Crash recovery : On application restart, afterSingletonsInstantiated reloads pending entries from Redis and resumes processing.