Netflix Hystrix笔记

maven引入

<dependency>

    <groupId>com.netflix.hystrix</groupId>

    <artifactId>hystrix-core</artifactId>

    <version>1.5.13</version>

</dependency>

hello,world

继承HystrixCommand；
HystrixCommandGroupKey是基类构造器必须的一个参数，可以实现接口来创建，也可以通过工厂方法HystrixCommandGroupKey.Factory.asKey(String)来创建；
实现run()方法；
调用

public class CommandHelloWorld extends HystrixCommand<String> {

    private final String name;

    public CommandHelloWorld(String name) {

        super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));

        this.name = name;

    }

    @Override

    protected String run() {

        // a real example would do work like a network call here

        return "Hello " + name + "!";

    }

}

@Test

public void testSynchronous() {

    assertEquals("Hello World!", new CommandHelloWorld("World").execute();

}

execute/queue/observe

String s1 = new CommandHelloWorld("World").execute();

Future<String> fs = new CommandHelloWorld("World").queue();

String s2 = fs.get();

Observable<String> ho = new CommandHelloWorld("World").observe();

ho.subscribe(new Action1<String>() {

    @Override

    public void call(String s) {

         // value emitted here

    }

});

Observable<String> co = new CommandHelloWorld("World").toObservable();

observe() — returns a “hot” Observable that executes the command immediately, though because the Observable is filtered through a ReplaySubject you are not in danger of losing any items that it emits before you have a chance to subscribe

toObservable() — returns a “cold” Observable that won’t execute the command and begin emitting its results until you subscribe to the Observable

Fallback

public class CommandHelloFailure extends HystrixCommand<String> {

    private final String name;

    public CommandHelloFailure(String name) {

        super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));

        this.name = name;

    }

    @Override

    protected String run() {

        throw new RuntimeException("this command always fails");

    }

    @Override

    protected String getFallback() {

        return "Hello Failure " + name + "!";

    }

}

Command Name & Command Group & Command Thread-Pool

public CommandHelloWorld(String name) {

    super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))

            .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld"))

            .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("HelloWorldPool")));

    this.name = name;

}

概念说明

Command Group: Hystrix uses the command group key to group together commands such as for reporting, alerting, dashboards, or team/library ownership.
Command Thread-Pool: The thread-pool key represents a HystrixThreadPool for monitoring, metrics publishing, caching, and other such uses. A HystrixCommand is associated with a single HystrixThreadPool as retrieved by the HystrixThreadPoolKey injected into it, or it defaults to one created using the HystrixCommandGroupKey it is created with.

缓存

通过实现getCacheKey()方法来开启缓存

public class CommandUsingRequestCache extends HystrixCommand<Boolean> {

    private final int value;

    protected CommandUsingRequestCache(int value) {

        super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));

        this.value = value;

    }

    @Override

    protected Boolean run() {

        return value == 0 || value % 2 == 0;

    }

    @Override

    protected String getCacheKey() {

        return String.valueOf(value);

    }

}

请求合并

request-scoped: 同一个HystrixRequestContext

globally-scoped: 跨多个HystrixRequestContext

关于请求合并这篇通过HystrixCollapser合并请求提高应用吞吐量的例子比较好，正好利用请求合并并发挥es的mget的作用。转载于此。

    //利用hystrix合并请求

    // scope:

    // 合并作用域，默认是REQUEST，就是不会跨越多个请求会话的，只在当前用户请求中合并多次请求为批处理请求。这里改成GLOBAL，就是可以跨越request context，合并不同用户的请求为一次批处理请求。

    // timerDelayInMilliseconds & maxRequestsInBatch:

    // 表示合并请求的时间窗口为50ms,窗口时间内最多合并200个请求。默认值是10ms，合并请求数不限。

    @HystrixCollapser(batchMethod = "mGetDataFromEs", scope = GLOBAL, collapserProperties = {

            @HystrixProperty(name = "timerDelayInMilliseconds", value = "50"),

            @HystrixProperty(name = "maxRequestsInBatch", value = "200"),

    })

    @Cacheable(value = "esRequestCache", cacheManager = "esRequestCacheManager", key = "#doc.index+':'+#doc.type+':'+#doc.id", unless = "#result == null || #result.size() ==0")

    public List<RecommendItem> getDataFromEs3(EsDocGetRequest doc) {

        return null;

    }

    // fallbackMethod: 指定fallback方法

    @HystrixCommand(fallbackMethod = "mGetDataFromEsFallback")

    public List<List<RecommendItem>> mGetDataFromEs(List<EsDocGetRequest> docs) throws IOException {

        logger.debug("================mGetDataFromEs==============");

        List<List<RecommendItem>> datas = new ArrayList<>();

        List<Doc> docList = new ArrayList<>();

        for (EsDocGetRequest req : docs) {

            docList.add(new Doc(req.getIndex(), req.getType(), req.getId()));

        }

        long start = System.currentTimeMillis();

        JestResult result = jestClient.execute(new MultiGetExt.Builder.ByDoc(docList).build());

        logger.info("============mGetDataFromEs cost time:{}", (System.currentTimeMillis() - start));

        // assertTrue(result.getErrorMessage(), result.isSucceeded());

        if (result.isSucceeded()) {

            JsonArray actualDocs = result.getJsonObject().getAsJsonArray("docs");

            for (int i = 0; i < docs.size(); i++) {

                boolean hasError = actualDocs.get(i).getAsJsonObject().has("error");

                boolean found = !hasError && actualDocs.get(i).getAsJsonObject().getAsJsonPrimitive("found").getAsBoolean();

                if (found) {

                    EsRcmdResult esRcmdResult = gson.fromJson(actualDocs.get(i).getAsJsonObject().get("_source"), EsRcmdResult.class);

                    datas.add(esRcmdResult.getData());

                } else {

                    datas.add(Collections.emptyList());

                }

            }

            return datas;

        } else {

            throw new ServiceException(result.getErrorMessage());

        }

    }

    public List<List<RecommendItem>> mGetDataFromEsFallback(List<EsDocGetRequest> docs) throws IOException {

        logger.error("============mGetDataFromEs fallback=============");

        return Collections.emptyList();

    }

Request Context Setup

public class HystrixRequestContextServletFilter implements Filter {

    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)

     throws IOException, ServletException {

        HystrixRequestContext context = HystrixRequestContext.initializeContext();

        try {

            chain.doFilter(request, response);

        } finally {

            context.shutdown();

        }

    }

}

Common Patterns

In the following sections are common uses and patterns of use for HystrixCommand and HystrixObservableCommand.

Fail Fast

has no fallback behivor
throws HystrixRuntimeException
通过HystrixRuntimeException.getCause获取原始异常对象

Fail Silent

Failing silently is the equivalent of returning an empty response or removing functionality. It can be done by returning null, an empty Map, empty List, or other such responses.

@Override

protected String getFallback() {

    return null;

}

@Override

protected List<String> getFallback() {

    return Collections.emptyList();

}

// HystrixObservableCommand

@Override

protected Observable<String> resumeWithFallback() {

    return Observable.empty();

}

Fallback: Static

在fallback方法里返回一个静态值

@Override

protected Boolean getFallback() {

    return true;

}

@Override

protected Observable<Boolean> resumeWithFallback() {

    return Observable.just( true );

}

Falllback: Stubbed

You typically use a stubbed fallback when your command returns a compound object containing multiple fields, some of which can be determined from other request state while other fields are set to default values.

Fallback: Cache via Network

Dynamic Property

example:primary--secondary-with-fallback

// write

ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", true);

// read

DynamicBooleanProperty usePrimary = DynamicPropertyFactory.getInstance().getBooleanProperty("primarySecondary.usePrimary", true);

Properties Strategy

If you implement a custom HystrixPropertiesStrategy, this gives you full control over how properties are defined for the system.

The default implementation uses Archaius.

todo: 如何运行时变更配置？

HystrixPlugins.getInstance().registerPropertiesStrategy(null);?

hystrix-javanica

什么是ReactiveX？

ReactiveX is a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data and/or events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety, concurrent data structures, and non-blocking I/O.

巴特西