{"id":32563,"date":"2014-11-11T07:00:10","date_gmt":"2014-11-11T05:00:10","guid":{"rendered":"http:\/\/www.javacodegeeks.com\/?p=32563"},"modified":"2014-11-10T07:07:27","modified_gmt":"2014-11-10T05:07:27","slug":"batching-collapsing-requests-in-hystrix","status":"publish","type":"post","link":"https:\/\/www.javacodegeeks.com\/2014\/11\/batching-collapsing-requests-in-hystrix.html","title":{"rendered":"Batching (collapsing) requests in Hystrix"},"content":{"rendered":"

Hystrix<\/a> has an advanced feature of collapsing (or batching)<\/a> requests. If two or more commands run similar request at the same time, Hystrix can combine them together, run one batched request and dispatch split results back to all commands. Let’s first see how Hystrix works without collapsing. Imagine we have a service that looks up StockPrice<\/code>of a given Ticker<\/code>:
\n 
\n 
\n 
\n 
\n 
\n <\/p>\n

import lombok.Value;\r\nimport java.math.BigDecimal;\r\nimport java.time.Instant;\r\n\r\n@Value\r\nclass Ticker {\r\n    String symbol;\r\n}\r\n\r\n@Value\r\nclass StockPrice {\r\n    BigDecimal price;\r\n    Instant effectiveTime;\r\n}\r\n\r\ninterface StockPriceGateway {\r\n\r\n    default StockPrice load(Ticker stock) {\r\n        final Set<Ticker> oneTicker = Collections.singleton(stock);\r\n        return loadAll(oneTicker).get(stock);\r\n    }\r\n\r\n    ImmutableMap<Ticker, StockPrice> loadAll(Set<Ticker> tickers);\r\n}\r\n<\/pre>\n
Core implementation of StockPriceGateway<\/code> must provide loadAll()<\/code> batch method while load()<\/code> method is implemented for our convenience. So our gateway is capable of loading multiple prices in one batch (e.g. to reduce latency or network protocol overhead), but at the moment we are not using this feature, always loading price of one stock at a time:<\/p>\n
class StockPriceCommand extends HystrixCommand<StockPrice> {\r\n\r\n    private final StockPriceGateway gateway;\r\n    private final Ticker stock;\r\n\r\n    StockPriceCommand(StockPriceGateway gateway, Ticker stock) {\r\n        super(HystrixCommandGroupKey.Factory.asKey(\"Stock\"));\r\n        this.gateway = gateway;\r\n        this.stock = stock;\r\n    }\r\n\r\n    @Override\r\n    protected StockPrice run() throws Exception {\r\n        return gateway.load(stock);\r\n    }\r\n}\r\n<\/pre>\n
Such command will always call StockPriceGateway.load()<\/code> for each and every Ticker<\/code>, as illustrated by the following tests:<\/p>\n
class StockPriceCommandTest extends Specification {\r\n\r\n    def gateway = Mock(StockPriceGateway)\r\n\r\n    def 'should fetch price from external service'() {\r\n        given:\r\n            gateway.load(TickerExamples.any()) >> StockPriceExamples.any()\r\n            def command = new StockPriceCommand(gateway, TickerExamples.any())\r\n\r\n        when:\r\n            def price = command.execute()\r\n\r\n        then:\r\n            price == StockPriceExamples.any()\r\n    }\r\n\r\n    def 'should call gateway exactly once when running Hystrix command'() {\r\n        given:\r\n            def command = new StockPriceCommand(gateway, TickerExamples.any())\r\n\r\n        when:\r\n            command.execute()\r\n\r\n        then:\r\n            1 * gateway.load(TickerExamples.any())\r\n    }\r\n\r\n    def 'should call gateway twice when command executed two times'() {\r\n        given:\r\n            def commandOne = new StockPriceCommand(gateway, TickerExamples.any())\r\n            def commandTwo = new StockPriceCommand(gateway, TickerExamples.any())\r\n\r\n        when:\r\n            commandOne.execute()\r\n            commandTwo.execute()\r\n\r\n        then:\r\n            2 * gateway.load(TickerExamples.any())\r\n    }\r\n\r\n    def 'should call gateway twice even when executed in parallel'() {\r\n        given:\r\n            def commandOne = new StockPriceCommand(gateway, TickerExamples.any())\r\n            def commandTwo = new StockPriceCommand(gateway, TickerExamples.any())\r\n\r\n        when:\r\n            Future<StockPrice> futureOne = commandOne.queue()\r\n            Future<StockPrice> futureTwo = commandTwo.queue()\r\n\r\n        and:\r\n            futureOne.get()\r\n            futureTwo.get()\r\n\r\n        then:\r\n            2 * gateway.load(TickerExamples.any())\r\n    }\r\n\r\n}\r\n<\/pre>\n
If you don’t know Hystrix, by wrapping an external call in a command you gain a lot of features like timeouts, circuit breakers, etc. But this is not the focus of this article. Look at last two tests: when asking for price of arbitrary ticker twice, sequentially or in parallel (queue()<\/code>), our external gateway<\/code> is also called twice. Last test is especially interesting – we ask for the same ticker at almost the same time, but Hystrix can’t figure that out. These two commands are fully independent, will be executed in different threads and don’t know anything about each other – even though they run at almost the same time.<\/p>\n
Collapsing is all about finding such similar requests and combining them. Batching (I will use this term interchangeably with collapsing<\/em>) doesn’t happen automatically and requires a bit of coding. But first let’s see how it behaves:<\/p>\n
def 'should collapse two commands executed concurrently for the same stock ticker'() {\r\n    given:\r\n        def anyTicker = TickerExamples.any()\r\n        def tickers = [anyTicker] as Set\r\n\r\n    and:\r\n        def commandOne = new StockTickerPriceCollapsedCommand(gateway, anyTicker)\r\n        def commandTwo = new StockTickerPriceCollapsedCommand(gateway, anyTicker)\r\n\r\n    when:\r\n        Future<StockPrice> futureOne = commandOne.queue()\r\n        Future<StockPrice> futureTwo = commandTwo.queue()\r\n\r\n    and:\r\n        futureOne.get()\r\n        futureTwo.get()\r\n\r\n    then:\r\n        0 * gateway.load(_)\r\n        1 * gateway.loadAll(tickers) >> ImmutableMap.of(anyTicker, StockPriceExamples.any())\r\n}\r\n\r\ndef 'should collapse two commands executed concurrently for the different stock tickers'() {\r\n    given:\r\n        def anyTicker = TickerExamples.any()\r\n        def otherTicker = TickerExamples.other()\r\n        def tickers = [anyTicker, otherTicker] as Set\r\n\r\n    and:\r\n        def commandOne = new StockTickerPriceCollapsedCommand(gateway, anyTicker)\r\n        def commandTwo = new StockTickerPriceCollapsedCommand(gateway, otherTicker)\r\n\r\n    when:\r\n        Future<StockPrice> futureOne = commandOne.queue()\r\n        Future<StockPrice> futureTwo = commandTwo.queue()\r\n\r\n    and:\r\n        futureOne.get()\r\n        futureTwo.get()\r\n\r\n    then:\r\n        1 * gateway.loadAll(tickers) >> ImmutableMap.of(\r\n                anyTicker, StockPriceExamples.any(),\r\n                otherTicker, StockPriceExamples.other())\r\n}\r\n\r\ndef 'should correctly map collapsed response into individual requests'() {\r\n    given:\r\n        def anyTicker = TickerExamples.any()\r\n        def otherTicker = TickerExamples.other()\r\n        def tickers = [anyTicker, otherTicker] as Set\r\n        gateway.loadAll(tickers) >> ImmutableMap.of(\r\n                anyTicker, StockPriceExamples.any(),\r\n                otherTicker, StockPriceExamples.other())\r\n\r\n    and:\r\n        def commandOne = new StockTickerPriceCollapsedCommand(gateway, anyTicker)\r\n        def commandTwo = new StockTickerPriceCollapsedCommand(gateway, otherTicker)\r\n\r\n    when:\r\n        Future<StockPrice> futureOne = commandOne.queue()\r\n        Future<StockPrice> futureTwo = commandTwo.queue()\r\n\r\n    and:\r\n        def anyPrice = futureOne.get()\r\n        def otherPrice = futureTwo.get()\r\n\r\n    then:\r\n        anyPrice == StockPriceExamples.any()\r\n        otherPrice == StockPriceExamples.other()\r\n}\r\n<\/pre>\n
First test proves that instead of calling load()<\/code> twice we barely called loadAll()<\/code> once. Also notice that since we asked for the same Ticker<\/code> (from two different threads), loadAll()<\/code> asks for only one ticker. Second test shows two concurrent requests for two different tickers being collapsed into one batch call. Third test makes sure we still get proper responses to each individual request. Instead of extending HystrixCommand<\/code> we must extend more complexHystrixCollapser<\/code>. Now it’s time to see StockTickerPriceCollapsedCommand<\/code> implementation, that seamlessly replaced StockPriceCommand<\/code>:
 
<\/div> <\/div><\/p>\n
class StockTickerPriceCollapsedCommand extends HystrixCollapser<ImmutableMap<Ticker, StockPrice>, StockPrice, Ticker> {\r\n\r\n    private final StockPriceGateway gateway;\r\n    private final Ticker stock;\r\n\r\n    StockTickerPriceCollapsedCommand(StockPriceGateway gateway, Ticker stock) {\r\n        super(HystrixCollapser.Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey(\"Stock\"))\r\n                .andCollapserPropertiesDefaults(HystrixCollapserProperties.Setter().withTimerDelayInMilliseconds(100)));\r\n        this.gateway = gateway;\r\n        this.stock = stock;\r\n    }\r\n\r\n    @Override\r\n    public Ticker getRequestArgument() {\r\n        return stock;\r\n    }\r\n\r\n    @Override\r\n    protected HystrixCommand<ImmutableMap<Ticker, StockPrice>> createCommand(Collection<CollapsedRequest<StockPrice, Ticker>> collapsedRequests) {\r\n        final Set<Ticker> stocks = collapsedRequests.stream()\r\n                .map(CollapsedRequest::getArgument)\r\n                .collect(toSet());\r\n        return new StockPricesBatchCommand(gateway, stocks);\r\n    }\r\n\r\n    @Override\r\n    protected void mapResponseToRequests(ImmutableMap<Ticker, StockPrice> batchResponse, Collection<CollapsedRequest<StockPrice, Ticker>> collapsedRequests) {\r\n        collapsedRequests.forEach(request -> {\r\n            final Ticker ticker = request.getArgument();\r\n            final StockPrice price = batchResponse.get(ticker);\r\n            request.setResponse(price);\r\n        });\r\n    }\r\n\r\n}\r\n<\/pre>\n
A lot is going on here, so let’s review StockTickerPriceCollapsedCommand<\/code> step by step. First three generic types:<\/p>\n