Hello World!
使用HystrixCommand实现“Hello World”。
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 + "!"; } }
使用HystrixObservableCommand实现“Hello World”。
public class CommandHelloWorld extends HystrixObservableCommand<String> { private final String name; public CommandHelloWorld(String name) { super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")); this.name = name; } @Override protected Observable<String> construct() { return Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> observer) { try { if (!observer.isUnsubscribed()) { // a real example would do work like a network call here observer.onNext("Hello"); observer.onNext(name + "!"); observer.onCompleted(); } } catch (Exception e) { observer.onError(e); } } } ).subscribeOn(Schedulers.io()); } }
同步执行
使用HystrixCommand的execute方法同步执行。
String s = new CommandHelloWorld("World").execute();
HystrixObservableCommand没有execute方法，但如果你确定Observable只会返回一个单一结果，你可以使用.toBlocking().toFuture().get()方法。
异步执行
使用HystrixCommand的queue()方法实现异步：
Future<String> fs = new CommandHelloWorld("World").queue();
通过get方式获得结果
String s = fs.get();
HystrixObservableCommand没有queue方法，但如果你确定Observable只会返回一个单一结果，你可以使用.toBlocking().toFuture()方法。
响应执行
通过下面两个方法可以获取Observable对象。
- observe，返回一个Observable并且立即执行命令，因为observe方法内部使用了一个RepaySubject，所以也会接受到监听以前命令返回的结果。
- toObservable，返回一个Observable但不回立即执行，当监听该对象时才执行。
Observable<String> ho = new CommandHelloWorld("World").observe(); // or Observable<String> co = new CommandHelloWorld("World").toObservable(); ho.subscribe(new Action1<String>() { @Override public void call(String s) { // value emitted here } });
使用单元测试执行
@Test public void testObservable() throws Exception { Observable<String> fWorld = new CommandHelloWorld("World").observe(); Observable<String> fBob = new CommandHelloWorld("Bob").observe(); // blocking assertEquals("Hello World!", fWorld.toBlockingObservable().single()); assertEquals("Hello Bob!", fBob.toBlockingObservable().single()); // non-blocking // - this is a verbose anonymous inner-class approach and doesn't do assertions fWorld.subscribe(new Observer<String>() { @Override public void onCompleted() { // nothing needed here } @Override public void onError(Throwable e) { e.printStackTrace(); } @Override public void onNext(String v) { System.out.println("onNext: " + v); } }); // non-blocking // - also verbose anonymous inner-class // - ignore errors and onCompleted signal fBob.subscribe(new Action1<String>() { @Override public void call(String v) { System.out.println("onNext: " + v); } }); }
响应式命令
除了使用上面的方式来获取Observable对象，还可以通过HystrixObservableCommand来创建一个Observable对象。HystrixObservableCommand可以创建一个返回多个值的Observable对象。HystrixObservableCommand使用consruct方法来执行命令而不是run方法。通过一下两种方式来获取Observable对象：
- observe，返回一个Observable并且立即执行命令，因为observe方法内部使用了一个RepaySubject，所以也会接受到监听以前命令返回的结果。
- toObservable，返回一个Observable但不回立即执行，当监听该对象时才执行。
降级
通过添加fallback方法可以让命令执行异常时执行降级措施。你会为大多数的hystrix 命令实现降级方法，除了一下的情况：
- 命令执行写操作
如果hystrix命令是被设计成一个写操作而非读操作（HystrixCommand命令返回void或者HystrixObservableCommand返回一个空的Observable对象）。那么没有什么实现fallback方法。如果写出错了，应该把异常抛出给调用方。
- 执行批量计算
如果hystrix命令使用了缓存、或者批量、或者离线技术，也应该把异常抛给调用方，让调用放过处理而不是降级。
无论你的命令是否有fallback，hystrix状态和熔断器状态都会更新。
在HystrixCommand中实现getFallback方法，当run方法失败、超时、线程池拒绝、信号量拒绝、熔断时将会执行getFallback，并返回结果。
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 + "!"; } }
run方法执行时将会抛出一个异常。但调用方会收到getFallback方法的返回结果。
使用HystrixObservableCommand的resumeWithFallback方法返回一个Observable对象也可以在执行命令失败后被调用返回。因为Observable对象可能在发射了一些数据之后才发生异常，所以可能在执行resumeWithFallback前返回数据。Hystrix内部使用了rxjava的onErrorResumeNext来实现。
异常传播
所有从run抛出的异常，除了HystrixBadRequestException都会执行getFallback方法并且进入熔断计算。你可以把你的异常封装成HystrixBadRequestException抛出，HystrixBadRequestException常常用于抛出参数错误，非系统异常不需要记入metric和执行降级的错误。
执行异常类型
| 失败类型 | 异常类 | 异常原因 |
| FAILURE | HystrixRuntimeException | underlying exception (user-controlled) |
| TIMEOUT | HystrixRuntimeException | TimeoutException |
| SHORT_CIRCUITED | HystrixRuntimeException | RuntimeException |
| THREAD_POOL_REJECTED | HystrixRuntimeException | RejectedExecutionException |
| SEMAPHORE_REJECTED | HystrixRuntimeException | RuntimeException |
| BAD_REQUEST | HystrixBadRequestException | underlying exception (user-controlled) |
命令名
命令名默认是使用类名
String name = cls.getSimpleName();
可以自定义命令名
public CommandHelloWorld(String name) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")) .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld"))); this.name = name; }
使用下面的方式来为每个command缓存Setter。
private static final Setter cachedSetter = Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")) .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld")); public CommandHelloWorld(String name) { super(cachedSetter); this.name = name; }
HystrixCommandKey是一个接口，并且提供了一个Factory来构造实现类。
HystrixCommandKey.Factory.asKey("HelloWorld")
组名
Hystrix使用组名来合并命令，用于报表，报警，项目管理。
如果没有指定线程池key，默认使用组名作为线程池key。
HystrixCommandGroupKey是一个接口，并且提供了一个Factory来构造实现类。
HystrixCommandGroupKey.Factory.asKey("ExampleGroup")
线程池key
线程池key唯一代表了一个线程池，用于监控、metrics、命令执行。默认使用HystrixCommandGroupKey。也可以显式指明：
public CommandHelloWorld(String name) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")) .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld")) .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("HelloWorldPool"))); this.name = name; }
HystrixThreadPoolKey是一个接口，并且提供了一个Factory来构造实现类。
HystrixThreadPoolKey.Factory.asKey("HelloWorldPool")
当多个命令属于同一个组，但是又要相互直接隔离时，可以显示指定HystrixThreadPoolKey。
缓存
通过实现HystrixCommand或HystrixObservableCommand的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); } }
因为请求缓存依赖HystrixRequestContext，所以在执行前需要先进行初始化。
@Test public void testWithoutCacheHits() { HystrixRequestContext context = HystrixRequestContext.initializeContext(); try { assertTrue(new CommandUsingRequestCache(2).execute()); assertFalse(new CommandUsingRequestCache(1).execute()); assertTrue(new CommandUsingRequestCache(0).execute()); assertTrue(new CommandUsingRequestCache(58672).execute()); } finally { context.shutdown(); } }
通常通过ServletFilter来初始化和关闭HystrixRequestContext。
@Test public void testWithCacheHits() { HystrixRequestContext context = HystrixRequestContext.initializeContext(); try { CommandUsingRequestCache command2a = new CommandUsingRequestCache(2); CommandUsingRequestCache command2b = new CommandUsingRequestCache(2); assertTrue(command2a.execute()); // this is the first time we've executed this command with // the value of "2" so it should not be from cache assertFalse(command2a.isResponseFromCache()); assertTrue(command2b.execute()); // this is the second time we've executed this command with // the same value so it should return from cache assertTrue(command2b.isResponseFromCache()); } finally { context.shutdown(); } // start a new request context context = HystrixRequestContext.initializeContext(); try { CommandUsingRequestCache command3b = new CommandUsingRequestCache(2); assertTrue(command3b.execute()); // this is a new request context so this // should not come from cache assertFalse(command3b.isResponseFromCache()); } finally { context.shutdown(); } }
请求合并
请求合并可以让多个请求合并成一到HystrixCommand中执行。可以设置合并请求数量和合并等待时间来触发请求合并。有两种作用域：请求级别和全局，通过构造函数指定，默认使用请求级别。
请求级别的合并通过一个HystrixRequestContext实现。全局的合并通过多个HystrixRequestContexts；如果依赖服务不能处理多个HystrixRequestContexts，最好使用请求级别作用域。
使用请求级别合并：
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++)); } } 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 { HystrixRequestContext context = HystrixRequestContext.initializeContext(); try { 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()); // 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 { context.shutdown(); } }
请求上下文
如果想要使用请求缓存、请求合并、请求日志的请求级别功能，我们必需管理HystrixRequestContext生命周期。
在请求开始之前执行下面的代码
HystrixRequestContext context = HystrixRequestContext.initializeContext();
在请求结束后执行下面的代码：
context.shutdown();
在java web应用中可以使用ServletFilter来管理HystrixRequestContext
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(); } } }
在web.xml中添加一下配置
<filter>
<display-name>HystrixRequestContextServletFilter</display-name>
<filter-name>HystrixRequestContextServletFilter</filter-name>
<filter-class>com.netflix.hystrix.contrib.requestservlet.HystrixRequestContextServletFilter</filter-class>
</filter>
<filter-mapping>
<filter-name>HystrixRequestContextServletFilter</filter-name>
<url-pattern>/*</url-pattern>
</filter-mapping>
公共特性
在下面的章节中，会讲HystrixCommand和HystrixObservableCommand的公共特性。
快速失败
大多数的使用是执行一个命令，并且没有fallback方法，如果碰到异常就会抛出
public class CommandThatFailsFast extends HystrixCommand<String> { private final boolean throwException; public CommandThatFailsFast(boolean throwException) { super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")); this.throwException = throwException; } @Override protected String run() { if (throwException) { throw new RuntimeException("failure from CommandThatFailsFast"); } else { return "success"; } }
单元测试
@Test public void testSuccess() { assertEquals("success", new CommandThatFailsFast(false).execute()); } @Test public void testFailure() { try { new CommandThatFailsFast(true).execute(); fail("we should have thrown an exception"); } catch (HystrixRuntimeException e) { assertEquals("failure from CommandThatFailsFast", e.getCause().getMessage()); e.printStackTrace(); } }
使用HystrixObservableCommand实现需要实现resumeWithFallback方法。
@Override protected Observable<String> resumeWithFallback() { if (throwException) { return Observable.error(new Throwable("failure from CommandThatFailsFast")); } else { return Observable.just("success"); } }
静默失败
静默失败是让fallback返回一个空结果。
public class CommandThatFailsSilently extends HystrixCommand<String> { private final boolean throwException; public CommandThatFailsSilently(boolean throwException) { super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")); this.throwException = throwException; } @Override protected String run() { if (throwException) { throw new RuntimeException("failure from CommandThatFailsFast"); } else { return "success"; } } @Override protected String getFallback() { return null; } }
单元测试
@Test public void testSuccess() { assertEquals("success", new CommandThatFailsSilently(false).execute()); } @Test public void testFailure() { try { assertEquals(null, new CommandThatFailsSilently(true).execute()); } catch (HystrixRuntimeException e) { fail("we should not get an exception as we fail silently with a fallback"); } }
另一种实现
@Override protected List<String> getFallback() { return Collections.emptyList(); }
使用HystrixObservableCommand实现
@Override protected Observable<String> resumeWithFallback() { return Observable.empty(); }
静态fallback
fallfack返回一个固定的静态值，例如如果命令失败返回true。
@Override protected Boolean getFallback() { return true; }
使用HystrixObservableCommand
@Override protected Observable<Boolean> resumeWithFallback() { return Observable.just( true ); }
存根fallback
通过咋fallback会返回一个包含请求信息和状态的对象，例如：cookies，请求参数，请求头，失败前service的返回结果。fallback可以通过请求作用域来获取信息，更典型的是在使用命令时通过构造函数设置这些值。
public class CommandWithStubbedFallback extends HystrixCommand<UserAccount> { private final int customerId; private final String countryCodeFromGeoLookup; /** * @param customerId * The customerID to retrieve UserAccount for * @param countryCodeFromGeoLookup * The default country code from the HTTP request geo code lookup used for fallback. */ protected CommandWithStubbedFallback(int customerId, String countryCodeFromGeoLookup) { super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")); this.customerId = customerId; this.countryCodeFromGeoLookup = countryCodeFromGeoLookup; } @Override protected UserAccount run() { // fetch UserAccount from remote service // return UserAccountClient.getAccount(customerId); throw new RuntimeException("forcing failure for example"); } @Override protected UserAccount getFallback() { /** * Return stubbed fallback with some static defaults, placeholders, * and an injected value 'countryCodeFromGeoLookup' that we'll use * instead of what we would have retrieved from the remote service. */ return new UserAccount(customerId, "Unknown Name", countryCodeFromGeoLookup, true, true, false); } public static class UserAccount { private final int customerId; private final String name; private final String countryCode; private final boolean isFeatureXPermitted; private final boolean isFeatureYPermitted; private final boolean isFeatureZPermitted; UserAccount(int customerId, String name, String countryCode, boolean isFeatureXPermitted, boolean isFeatureYPermitted, boolean isFeatureZPermitted) { this.customerId = customerId; this.name = name; this.countryCode = countryCode; this.isFeatureXPermitted = isFeatureXPermitted; this.isFeatureYPermitted = isFeatureYPermitted; this.isFeatureZPermitted = isFeatureZPermitted; } } }
单元测试
@Test public void test() { CommandWithStubbedFallback command = new CommandWithStubbedFallback(1234, "ca"); UserAccount account = command.execute(); assertTrue(command.isFailedExecution()); assertTrue(command.isResponseFromFallback()); assertEquals(1234, account.customerId); assertEquals("ca", account.countryCode); assertEquals(true, account.isFeatureXPermitted); assertEquals(true, account.isFeatureYPermitted); assertEquals(false, account.isFeatureZPermitted); }
使用HystrixObservableCommand
@Override protected Observable<Boolean> resumeWithFallback() { return Observable.just( new UserAccount(customerId, "Unknown Name", countryCodeFromGeoLookup, true, true, false) ); }
如果使用返回多个数据，那么需要知道在异常前返回了哪些数据，然后在fallback继续处理。
@Override protected Observable<Integer> construct() { return Observable.just(1, 2, 3) .concatWith(Observable.<Integer> error(new RuntimeException("forced error"))) .doOnNext(new Action1<Integer>() { @Override public void call(Integer t1) { lastSeen = t1; } }) .subscribeOn(Schedulers.computation()); } @Override protected Observable<Integer> resumeWithFallback() { if (lastSeen < 4) { return Observable.range(lastSeen + 1, 4 - lastSeen); } else { return Observable.empty(); } }
缓存fallback
有时候，如果执行异常需要在缓存中获取数据来代替。因为从缓存中获取数据也可能需要网络请求，所以也需要通过HystrixCommand或HystrixObservableCommand来处理
fallback使用的线程池必须和调用方法的隔离开，如果隔离开，那么当调用服务超负载是，fallback也会不可执行。下面的例子显示了CommandWithFallbackViaNetwork如何在他的getFallback中执行FallbackViaNetwork。如果FallbackViaNetwork执行失败了，他也有自己的一个fallback，返回null。
为了让FallbackViaNetwork和CommandWithFallbackViaNetwork不在同一个线程池中执行，通过HystrixThreadPoolKey.Factory.asKey("RemoteServiceXFallback")设置了他自己的线程池。
public class CommandWithFallbackViaNetwork extends HystrixCommand<String> { private final int id; protected CommandWithFallbackViaNetwork(int id) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("RemoteServiceX")) .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueCommand"))); this.id = id; } @Override protected String run() { // RemoteServiceXClient.getValue(id); throw new RuntimeException("force failure for example"); } @Override protected String getFallback() { return new FallbackViaNetwork(id).execute(); } private static class FallbackViaNetwork extends HystrixCommand<String> { private final int id; public FallbackViaNetwork(int id) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("RemoteServiceX")) .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueFallbackCommand")) // use a different threadpool for the fallback command // so saturating the RemoteServiceX pool won't prevent // fallbacks from executing .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("RemoteServiceXFallback"))); this.id = id; } @Override protected String run() { MemCacheClient.getValue(id); } @Override protected String getFallback() { // the fallback also failed // so this fallback-of-a-fallback will // fail silently and return null return null; } } }
一些系统有主备模式，一般备用模式只用在失败的情况下，这种模式就是前一个例子讲的情况。然后自动切换主备并不是一种理想的方案，当主模式失败时，他会忽略报警。
为了解决这个问题，可以使用手动切换主备模式。
主备的HystrixCommand实现是相互线程隔离的，他们有不同的表现特性。
public class CommandFacadeWithPrimarySecondary extends HystrixCommand<String> { private final static DynamicBooleanProperty usePrimary = DynamicPropertyFactory.getInstance().getBooleanProperty("primarySecondary.usePrimary", true); private final int id; public CommandFacadeWithPrimarySecondary(int id) { super(Setter .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX")) .andCommandKey(HystrixCommandKey.Factory.asKey("PrimarySecondaryCommand")) .andCommandPropertiesDefaults( // we want to default to semaphore-isolation since this wraps // 2 others commands that are already thread isolated HystrixCommandProperties.Setter() .withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE))); this.id = id; } @Override protected String run() { if (usePrimary.get()) { return new PrimaryCommand(id).execute(); } else { return new SecondaryCommand(id).execute(); } } @Override protected String getFallback() { return "static-fallback-" + id; } @Override protected String getCacheKey() { return String.valueOf(id); } private static class PrimaryCommand extends HystrixCommand<String> { private final int id; private PrimaryCommand(int id) { super(Setter .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX")) .andCommandKey(HystrixCommandKey.Factory.asKey("PrimaryCommand")) .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("PrimaryCommand")) .andCommandPropertiesDefaults( // we default to a 600ms timeout for primary HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(600))); this.id = id; } @Override protected String run() { // perform expensive 'primary' service call return "responseFromPrimary-" + id; } } private static class SecondaryCommand extends HystrixCommand<String> { private final int id; private SecondaryCommand(int id) { super(Setter .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX")) .andCommandKey(HystrixCommandKey.Factory.asKey("SecondaryCommand")) .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("SecondaryCommand")) .andCommandPropertiesDefaults( // we default to a 100ms timeout for secondary HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(100))); this.id = id; } @Override protected String run() { // perform fast 'secondary' service call return "responseFromSecondary-" + id; } } public static class UnitTest { @Test public void testPrimary() { HystrixRequestContext context = HystrixRequestContext.initializeContext(); try { ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", true); assertEquals("responseFromPrimary-20", new CommandFacadeWithPrimarySecondary(20).execute()); } finally { context.shutdown(); ConfigurationManager.getConfigInstance().clear(); } } @Test public void testSecondary() { HystrixRequestContext context = HystrixRequestContext.initializeContext(); try { ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", false); assertEquals("responseFromSecondary-20", new CommandFacadeWithPrimarySecondary(20).execute()); } finally { context.shutdown(); ConfigurationManager.getConfigInstance().clear(); } } } }
无网络请求的依赖
如果执行的依赖不需要网络请求，可以设置executionIsolationStrategy为ExecutionIsolationStrategy.SEMAPHORE通过信号量来隔离。
public class CommandUsingSemaphoreIsolation extends HystrixCommand<String> { private final int id; public CommandUsingSemaphoreIsolation(int id) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")) // since we're doing an in-memory cache lookup we choose SEMAPHORE isolation .andCommandPropertiesDefaults(HystrixCommandProperties.Setter() .withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE))); this.id = id; } @Override protected String run() { // a real implementation would retrieve data from in memory data structure return "ValueFromHashMap_" + id; } }
清除缓存
如果当数据改变时，需要清除缓存，可以通过HystrixRequestCache.clear()来清除。
public class CommandUsingRequestCacheInvalidation { /* represents a remote data store */ private static volatile String prefixStoredOnRemoteDataStore = "ValueBeforeSet_"; public static class GetterCommand extends HystrixCommand<String> { 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. * * @param id * argument that would normally be passed to the command */ public static void flushCache(int id) { HystrixRequestCache.getInstance(GETTER_KEY, HystrixConcurrencyStrategyDefault.getInstance()).clear(String.valueOf(id)); } } public static class SetterCommand extends HystrixCommand<Void> { private final int id; private final String prefix; public SetterCommand(int id, String prefix) { super(HystrixCommandGroupKey.Factory.asKey("GetSetGet")); this.id = id; this.prefix = prefix; } @Override protected Void run() { // persist the value against the datastore prefixStoredOnRemoteDataStore = prefix; // flush the cache GetterCommand.flushCache(id); // no return value return null; } } }
单元测试
@Test public void getGetSetGet() { HystrixRequestContext context = HystrixRequestContext.initializeContext(); try { assertEquals("ValueBeforeSet_1", new GetterCommand(1).execute()); GetterCommand commandAgainstCache = new GetterCommand(1); assertEquals("ValueBeforeSet_1", commandAgainstCache.execute()); // confirm it executed against cache the second time assertTrue(commandAgainstCache.isResponseFromCache()); // set the new value new SetterCommand(1, "ValueAfterSet_").execute(); // fetch it again GetterCommand commandAfterSet = new GetterCommand(1); // the getter should return with the new prefix, not the value from cache assertFalse(commandAfterSet.isResponseFromCache()); assertEquals("ValueAfterSet_1", commandAfterSet.execute()); } finally { context.shutdown(); } } }
如果想使用Hystrix，那么需要将依赖服务调用通过HystrixCommand来封装。
使用Hystrix之前的结构
使用Hystrix之后的结构
