Request Collapsing Demystified: Hystrix, BatchCollapser & ConcurrentHashMultiset

Introduction

Combining similar or duplicate requests upstream can greatly reduce downstream load and improve overall system throughput. This article introduces three request‑merging techniques—Hystrix Collapser, a custom BatchCollapser, and Guava's ConcurrentHashMultiset—and compares their suitable scenarios.

1. Hystrix Collapser

Hystrix is an open‑source library from Netflix that provides circuit breaking and request collapsing. It is often used via the javanica module with annotations, requiring hystrix-core and hystrix-javanica dependencies and an AOP configuration.

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

Hystrix Collapser can be defined with @HystrixCollapser (specifying batchMethod) and @HystrixCommand on the batch method. Important points:

Add @HystrixCollapser on the method to be collapsed and @HystrixCommand on the batch method.

The single method accepts a single parameter; for multiple parameters wrap them in a class. The batch method receives a List of the single‑parameter type. single returns Future<T>; batch returns List<T> with the same order as input.

Simple example:

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

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

Configuration of Hystrix Collapser is done via annotation attributes:

collapserKey (optional, defaults to method name)

batchMethod – name of the batch method

scope – REQUEST or GLOBAL (GLOBAL creates a single collapser per JVM)

collapserProperties – common HystrixCommand properties such as maxRequestsInBatch, timerDelayInMilliseconds,

requestCache.enabled

@HystrixCollapser(
    batchMethod = "batch",
    collapserKey = "single",
    scope = com.netflix.hystrix.HystrixCollapser.Scope.GLOBAL,
    collapserProperties = {
        @HystrixProperty(name = "maxRequestsInBatch", value = "100"),
        @HystrixProperty(name = "timerDelayInMilliseconds", value = "1000"),
        @HystrixProperty(name = "requestCache.enabled", value = "true")
    })

Note: Because the collapser waits for the timer thread to trigger the batch request, each request incurs an additional cost of roughly timerInterval/2 ms.

2. Custom BatchCollapser

Inspired by Hystrix, a simple BatchCollapser was implemented. Business threads submit requests to a container; when the number of pending requests reaches a threshold or a time interval elapses, a timer thread consumes the container and sends a combined request downstream.

Key design points:

Singleton implementation using a concurrent map and double‑checked locking.

Factory receives a Handler implementation to group requests by handler class. ScheduledExecutorService is used instead of java.util.Timer to avoid blocking issues.

Implementation (simplified):

public class BatchCollapser<E> implements InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(BatchCollapser.class);
    private static volatile Map<Class, BatchCollapser> instance = Maps.newConcurrentMap();
    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 getInstance(Handler<List<E>, Boolean> cleaner, int threshHold, long interval) {
        Class jobClass = cleaner.getClass();
        if (instance.get(jobClass) == null) {
            synchronized (BatchCollapser.class) {
                if (instance.get(jobClass) == null) {
                    instance.put(jobClass, new BatchCollapser<>(cleaner, threshHold, interval));
                }
            }
        }
        return instance.get(jobClass);
    }
}

3. ConcurrentHashMultiset

Guava's ConcurrentHashMultiset stores elements with a count, allowing thread‑safe incrementing of duplicate items without locking. It is well‑suited for high‑frequency statistical scenarios where many identical requests arrive.

Typical usage merges requests by counting duplicates and then sending a single aggregated request downstream:

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);
});

Conclusion

Suitable scenarios:

Hystrix Collapser – when each request’s result is needed and extra latency is acceptable.

BatchCollapser – when results are not required and merging can be triggered by time or count.

ConcurrentHashMultiset – for high‑duplicate statistical workloads.

Combining BatchCollapser with ConcurrentHashMultiset can further leverage both time‑based merging and duplicate counting.