Hystrix框架5--請求緩存和collapser

簡介

在Hystrix中有個Request的概念，有一些操作需要在request中進行

緩存

在Hystrix調用服務時，如果只是查詢接口，可以使用緩存進行優化，從而跳過真實訪問請求。

應用

需要啟用緩存的話需要重寫command中getCacheKey方法

    @Override
    protected String getCacheKey() {
        return String.valueOf(value);
    }

之后就可以調用了
但是如果直接調用command的運行相關方法會得到以下錯誤

Caused by: java.lang.IllegalStateException: Request caching is not available. Maybe you need to initialize the HystrixRequestContext?

這里說需要初始化HystrixRequestContext，在Hystrix中帶緩存的command是必須在request中調用的。

public void testWithCacheHits() {
	//初始化context
    HystrixRequestContext context = HystrixRequestContext.initializeContext();
    try {
		//兩個一樣的command
        CommandUsingRequestCache command2a = new CommandUsingRequestCache(2);
        CommandUsingRequestCache command2b = new CommandUsingRequestCache(2);
		
        assertTrue(command2a.execute());
        // 第一次調用的返回是否從cache獲得是false
        assertFalse(command2a.isResponseFromCache());

        assertTrue(command2b.execute());
        // 第而次調用的返回是否從cache獲得是true
        assertTrue(command2b.isResponseFromCache());
    } finally {
		//關閉context
        context.shutdown();
    }
}

在不同context中的緩存是不共享的，還有這個request內部一個ThreadLocal，所以request只能限於當前線程

更新緩存

如果服務同時有更新的操作，那么在request中需要將原先的緩存失效

public static class GetterCommand extends HystrixCommand<String> {
	//getter key 用於命名當前command，之后清楚緩存時需要
    private static final HystrixCommandKey GETTER_KEY = HystrixCommandKey.Factory.asKey("GetterCommand");
    private final int id;

    public GetterCommand(int id) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("GetSetGet"))
                .andCommandKey(GETTER_KEY));
        this.id = id;
    }

    @Override
    protected String run() {
        return prefixStoredOnRemoteDataStore + id;
    }

    @Override
    protected String getCacheKey() {
        return String.valueOf(id);
    }

    /**
     * Allow the cache to be flushed for this object.
     */
    public static void flushCache(int id) {
		//這里從HystrixRequestCache的getInstance靜態方法中找到對應實例，並將響應值清除
        HystrixRequestCache.getInstance(GETTER_KEY,
                HystrixConcurrencyStrategyDefault.getInstance()).clear(String.valueOf(id));
    }
}

Collapser

Collapser可以合並多個請求一起調用

//需要定義一個Collapser
public class CommandCollapserGetValueForKey extends HystrixCollapser<List<String>, String, Integer> {

    private final Integer key;

    public CommandCollapserGetValueForKey(Integer key) {
        this.key = key;
    }

    @Override
    public Integer getRequestArgument() {
        return key;
    }

    @Override
    protected HystrixCommand<List<String>> createCommand(final Collection<CollapsedRequest<String, Integer>> requests) {
        return new BatchCommand(requests);
    }

    @Override
    protected void mapResponseToRequests(List<String> batchResponse, Collection<CollapsedRequest<String, Integer>> requests) {
        int count = 0;
        for (CollapsedRequest<String, Integer> request : requests) {
            request.setResponse(batchResponse.get(count++));
        }
    }
	//底層command實現
    private static final class BatchCommand extends HystrixCommand<List<String>> {
        private final Collection<CollapsedRequest<String, Integer>> requests;

        private BatchCommand(Collection<CollapsedRequest<String, Integer>> requests) {
                super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
                    .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueForKey")));
            this.requests = requests;
        }
		//內部還是遍歷入參進行調用
        @Override
        protected List<String> run() {
            ArrayList<String> response = new ArrayList<String>();
            for (CollapsedRequest<String, Integer> request : requests) {
                // artificial response for each argument received in the batch
                response.add("ValueForKey: " + request.getArgument());
            }
            return response;
        }
    }
}

@Test
public void testCollapser() throws Exception {
    //初始化request
    HystrixRequestContext context = HystrixRequestContext.initializeContext();
    try {
		//直接調用collapser
        Future<String> f1 = new CommandCollapserGetValueForKey(1).queue();
        Future<String> f2 = new CommandCollapserGetValueForKey(2).queue();
        Future<String> f3 = new CommandCollapserGetValueForKey(3).queue();
        Future<String> f4 = new CommandCollapserGetValueForKey(4).queue();

        assertEquals("ValueForKey: 1", f1.get());
        assertEquals("ValueForKey: 2", f2.get());
        assertEquals("ValueForKey: 3", f3.get());
        assertEquals("ValueForKey: 4", f4.get());
		//只調用了一次command
        // assert that the batch command 'GetValueForKey' was in fact
        // executed and that it executed only once
        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
        HystrixCommand<?> command = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand<?>[1])[0];
        // assert the command is the one we're expecting
        assertEquals("GetValueForKey", command.getCommandKey().name());
        // confirm that it was a COLLAPSED command execution
        assertTrue(command.getExecutionEvents().contains(HystrixEventType.COLLAPSED));
        // and that it was successful
        assertTrue(command.getExecutionEvents().contains(HystrixEventType.SUCCESS));
    } finally {
        //關閉request
        context.shutdown();
    }
}

Collapser可以用來一起調用一批請求，在第一個調用get時會對積壓的command一起調用。

總結

Hystrix的Request主要就是cache和collapser用到。在一般web應用中可以在filter中加入context的初始化和關閉邏輯。還要注意request是不支持跨線程。