Mastering Request Collapsing: Hystrix, Custom BatchCollapser & ConcurrentHashMultiset

Preface

In many services the request‑response model creates a separate thread and memory space for each call, making I/O operations costly. Merging similar requests before they reach downstream systems can greatly lower the load on those services and improve overall system throughput.

Hystrix Collapser

Hystrix, an open‑source library from Netflix, provides a collapser that can batch requests. It works together with the HystrixCommand and is often used via the Javanica module, which allows annotation‑based definitions.

<aop:aspectj-autoproxy/>
<bean id="hystrixAspect" class="com.netflix.hystrix.contrib.javanica.aop.aspectj.HystrixCommandAspect" />

To enable request collapsing you annotate the single request method with @HystrixCollapser (specifying the batch method) and the batch method with @HystrixCommand. Important points:

The single method can only accept one parameter; for multiple parameters you must wrap them in a custom class.

The single method returns Future<SingleReturn>, while the batch method returns List<SingleReturn>, and the result count must match the input count.

Example:

public class HystrixCollapserSample {
    @HystrixCollapser(batchMethod = "batch")
    public Future<Boolean> single(String input) {
        return null; // never executed directly
    }

    public List<Boolean> batch(List<String> inputs) {
        return inputs.stream().map(it -> Boolean.TRUE).collect(Collectors.toList());
    }
}

The collapser registers an AOP bean, detects the annotation at runtime, creates a collapser instance, stores request arguments in a concurrent map, and uses a timer thread to batch and execute the combined request.

BatchCollapser (Custom Implementation)

When the result of each request is not needed, a lightweight custom collapser can be built. It stores incoming requests in a container, and a timer thread triggers batch execution either after a time interval or when a request count threshold is reached.

public class BatchCollapser<E> implements InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(BatchCollapser.class);
    private static final ScheduledExecutorService SCHEDULE_EXECUTOR = Executors.newScheduledThreadPool(1);
    private volatile LinkedBlockingDeque<E> batchContainer = new LinkedBlockingDeque<>();
    private Handler<List<E>, Boolean> cleaner;
    private long interval;
    private int threshHold;

    private BatchCollapser(Handler<List<E>, Boolean> cleaner, int threshHold, long interval) {
        this.cleaner = cleaner;
        this.threshHold = threshHold;
        this.interval = interval;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        SCHEDULE_EXECUTOR.scheduleAtFixedRate(() -> {
            try { clean(); } catch (Exception e) { logger.error("clean container exception", e); }
        }, 0, interval, TimeUnit.MILLISECONDS);
    }

    public void submit(E event) {
        batchContainer.add(event);
        if (batchContainer.size() >= threshHold) { clean(); }
    }

    private void clean() {
        List<E> transferList = Lists.newArrayListWithExpectedSize(threshHold);
        batchContainer.drainTo(transferList, 100);
        if (CollectionUtils.isEmpty(transferList)) { return; }
        try { cleaner.handle(transferList); } catch (Exception e) { logger.error("batch execute error, transferList:{}", transferList, e); }
    }

    public static <E> BatchCollapser<E> getInstance(Handler<List<E>, Boolean> cleaner, int threshHold, long interval) {
        // singleton factory omitted for brevity
    }
}

The design mirrors Hystrix but returns immediately after submitting the request, allowing the timer or count threshold to trigger the actual batch execution.

ConcurrentHashMultiset (Guava)

For scenarios with high request duplication, Guava's ConcurrentHashMultiset provides a thread‑safe multiset that counts occurrences of each element without locking. It is ideal for in‑memory aggregation before persisting to a database.

if (ConcurrentHashMultiset.isEmpty()) {
    return;
}

List<Request> transferList = Lists.newArrayList();
ConcurrentHashMultiset.elementSet().forEach(request -> {
    int count = ConcurrentHashMultiset.count(request);
    if (count <= 0) { return; }
    transferList.add(count == 1 ? request : new Request(request.getIncrement() * count));
    ConcurrentHashMultiset.remove(request, count);
});

Summary of Suitable Scenarios

hystrix collapser

: when each request’s result is required and the extra cost of batching is acceptable. BatchCollapser: when results are not needed and you want batching triggered by time or count thresholds. ConcurrentHashMultiset: when the workload involves high‑frequency duplicate requests, such as statistical counting.

By combining these techniques—using a custom BatchCollapser together with ConcurrentHashMultiset as the container—you can achieve both time‑based and count‑based batching while keeping memory usage low and thread safety guaranteed.