hystrix進行資源隔離，其實是提供了一個抽象，叫做command，就是說，你如果要把對某一個依賴服務的所有調用請求，全部隔離在同一份資源池內

對這個依賴服務的所有調用請求，全部走這個資源池內的資源，不會去用其他的資源了，這個就叫做資源隔離

hystrix最最基本的資源隔離的技術，線程池隔離技術

對某一個依賴服務，商品服務，所有的調用請求，全部隔離到一個線程池內，對商品服務的每次調用請求都封裝在一個command里面

每個command(每次服務調用請求)都是使用線程池內的一個線程去執行的

所以哪怕是對這個依賴服務，商品服務，現在同時發起的調用量已經到了1000了，但是線程池內就10個線程，最多就只會用這10個線程去執行

不會說，對商品服務的請求，因為接口調用延遲，將tomcat內部所有的線程資源全部耗盡，不會出現了

1.pox

com.netflix.hystrix

hystrix-core

1.5.12

import rx.Observable;

import rx.Subscriber;

import rx.schedulers.Schedulers;

import com.alibaba.fastjson.JSONObject;

import com.netflix.hystrix.HystrixCommandGroupKey;

import com.netflix.hystrix.HystrixObservableCommand;

import com.roncoo.eshop.cache.ha.http.HttpClientUtils;

import com.roncoo.eshop.cache.ha.model.ProductInfo;

public class GetProductInfosCommand extends HystrixObservableCommand {

private String[] productIds;

public GetProductInfosCommand(String[] productIds) {

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

this.productIds = productIds;

}

@Override

protected Observable construct() {

return Observable.create(new Observable.OnSubscribe() {

public void call(Subscriber super ProductInfo> observer) {

try {

for(String productId : productIds) {

String url = "http://127.0.0.1:8082/getProductInfo?productId=" + productId;

String response = HttpClientUtils.sendGetRequest(url);

ProductInfo productInfo = JSONObject.parseObject(response, ProductInfo.class);

observer.onNext(productInfo);

}

observer.onCompleted();

} catch (Exception e) {

observer.onError(e);

}

}

}).subscribeOn(Schedulers.io());

}

}

import com.alibaba.fastjson.JSONObject;

import com.netflix.hystrix.HystrixCommand;

import com.netflix.hystrix.HystrixCommandGroupKey;

import com.roncoo.eshop.cache.ha.http.HttpClientUtils;

import com.roncoo.eshop.cache.ha.model.ProductInfo;

public class GetProductInfoCommand extends HystrixCommand {

private Long productId;

public GetProductInfoCommand(Long productId) {

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

this.productId = productId;

}

@Override

protected ProductInfo run() throws Exception {

String url = "http://127.0.0.1:8082/getProductInfo?productId=" + productId;

String response = HttpClientUtils.sendGetRequest(url);

return JSONObject.parseObject(response, ProductInfo.class);

}

}

import org.springframework.stereotype.Controller;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.ResponseBody;

import rx.Observable;

import rx.Observer;

import com.netflix.hystrix.HystrixCommand;

import com.netflix.hystrix.HystrixObservableCommand;

import com.roncoo.eshop.cache.ha.http.HttpClientUtils;

import com.roncoo.eshop.cache.ha.hystrix.command.GetProductInfoCommand;

import com.roncoo.eshop.cache.ha.hystrix.command.GetProductInfosCommand;

import com.roncoo.eshop.cache.ha.model.ProductInfo;

@Controller

public class CacheController {

@RequestMapping("/change/product")

@ResponseBody

public String changeProduct(Long productId) {

// 拿到一個商品id

// 調用商品服務的接口，獲取商品id對應的商品的最新數據

// 用HttpClient去調用商品服務的http接口

String url = "http://127.0.0.1:8082/getProductInfo?productId=" + productId;

String response = HttpClientUtils.sendGetRequest(url);

//Future future = getProductInfoCommand.queue();

//try {

//Thread.sleep(1000);

//System.out.println(future.get());

//} catch (Exception e) {

//e.printStackTrace();

//}

System.out.println(response);

return "success";

}

/**

* nginx開始，各級緩存都失效了，nginx發送很多的請求直接到緩存服務要求拉取最原始的數據

*

* @param productId

* @return

*/

@RequestMapping("/getProductInfo")

@ResponseBody

public String getProductInfo(Long productId) {

// 拿到一個商品id

// 調用商品服務的接口，獲取商品id對應的商品的最新數據

// 用HttpClient去調用商品服務的http接口

HystrixCommand getProductInfoCommand = new GetProductInfoCommand(productId);

ProductInfo productInfo = getProductInfoCommand.execute();

System.out.println(productInfo);

return "success";

}

/**

* 一次性批量查詢多條商品數據的請求

*/

@RequestMapping("/getProductInfos")

@ResponseBody

public String getProductInfos(String productIds) {

HystrixObservableCommand getProductInfosCommand =

new GetProductInfosCommand(productIds.split(","));

Observable observable = getProductInfosCommand.observe();

//observable = getProductInfosCommand.toObservable(); // 還沒有執行

observable.subscribe(new Observer() { // 等到調用subscribe然后才會執行

public void onCompleted() {

System.out.println("獲取完了所有的商品數據");

}

public void onError(Throwable e) {

e.printStackTrace();

}

public void onNext(ProductInfo productInfo) {

System.out.println(productInfo);

}

});

return "success";

}

}

command的四種調用方式

同步：new CommandHelloWorld("World").execute()，new ObservableCommandHelloWorld("World").toBlocking().toFuture().get()

如果你認為observable command只會返回一條數據，那么可以調用上面的模式，去同步執行，返回一條數據

異步：new CommandHelloWorld("World").queue()，new ObservableCommandHelloWorld("World").toBlocking().toFuture()

對command調用queue()，僅僅將command放入線程池的一個等待隊列，就立即返回，拿到一個Future對象，后面可以做一些其他的事情，然后過一段時間對future調用get()方法獲取數據。

1.線程池隔離技術與信號量隔離技術的區別

hystrix里面，核心的一項功能，其實就是所謂的資源隔離，要解決的最最核心的問題，就是將多個依賴服務的調用分別隔離到各自自己的資源池內

避免說對某一個依賴服務的調用，因為依賴服務的接口調用的延遲或者失敗，導致服務所有的線程資源全部耗費在這個服務的接口調用上

一旦說某個服務的線程資源全部耗盡的話，可能就導致服務就會崩潰，甚至說這種故障會不斷蔓延

hystrix，資源隔離，兩種技術，線程池的資源隔離，信號量的資源隔離

信號量，semaphore

信號量：適合，你的訪問不是對外部依賴的訪問，而是對內部的一些比較復雜的業務邏輯的訪問，但是像這種訪問，系統內部的代碼，其實不涉及任何的網絡請求，

那么只要做信號量的普通限流就可以了，因為不需要去捕獲timeout類似的問題，算法+數據結構的效率不是太高，

并發量突然太高，因為這里稍微耗時一些，導致很多線程卡在這里的話，不太好，

所以進行一個基本的資源隔離和訪問，避免內部復雜的低效率的代碼，導致大量的線程被卡住。

一般我們在獲取到商品數據之后，都要去獲取商品是屬于哪個地理位置，省，市，賣家的，可能在自己的純內存中，比如就一個Map去獲取

對于這種直接訪問本地內存的邏輯，比較適合用信號量做一下簡單的隔離

優點在于，不用自己管理線程池，不用care timeout超時了，信號量做隔離的話，性能會相對來說高一些。

import com.netflix.hystrix.HystrixCommand;

import com.netflix.hystrix.HystrixCommandGroupKey;

import com.netflix.hystrix.HystrixCommandProperties;

import com.netflix.hystrix.HystrixCommandProperties.ExecutionIsolationStrategy;

import com.roncoo.eshop.cache.ha.local.LocationCache;

public class GetCityNameCommand extends HystrixCommand {

private Long cityId;

public GetCityNameCommand(Long cityId){

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

.andCommandPropertiesDefaults(HystrixCommandProperties.Setter()

.withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)));

this.cityId = cityId;

}

@Override

protected String run() throws Exception {

return LocationCache.getCityName(cityId);

}

}

execution.isolation.strategy：

指定了HystrixCommand.run()的資源隔離策略，THREAD或者SEMAPHORE，一種是基于線程池，一種是信號量

線程池機制，每個command運行在一個線程中，限流是通過線程池的大小來控制的

信號量機制，command是運行在調用線程中，但是通過信號量的容量來進行限流

如何在線程池和信號量之間做選擇？

默認的策略就是線程池

而使用信號量的場景，通常是針對超大并發量的場景下，每個服務實例每秒都幾百的QPS，那么此時你用線程池的話，線程一般不會太多，

可能撐不住那么高的并發，如果要撐住，可能要耗費大量的線程資源，那么就是用信號量，來進行限流保護

一般用信號量常見于那種基于純內存的一些業務邏輯服務，而不涉及到任何網絡訪問請求

netflix有100+的command運行在40+的線程池中，只有少數command是不運行在線程池中的，就是從純內存中獲取一些元數據，或者是對多個command包裝起來的facacde command，是用信號量限流的.

// to use thread isolation

HystrixCommandProperties.Setter()

.withExecutionIsolationStrategy(ExecutionIsolationStrategy.THREAD)

// to use semaphore isolation

HystrixCommandProperties.Setter()

.withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)

2、command名稱和command組

每個command，都可以設置一個自己的名稱，同時可以設置一個自己的組.

command group，是一個非常重要的概念，默認情況下，因為就是通過command group來定義一個線程池的，而且還會通過command group來聚合一些監控和報警信息

同一個command group中的請求，都會進入同一個線程池中

3、command線程池

threadpool key代表了一個HystrixThreadPool，用來進行統一監控，統計，緩存

默認的threadpool key就是command group名稱

每個command都會跟它的threadpool key對應的thread pool綁定在一起

如果不想直接用command group，也可以手動設置thread pool name

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 threadpool -> command group -> command key

command key，代表了一類command，一般來說，代表了底層的依賴服務的一個接口

command group，代表了某一個底層的依賴服務，合理，一個依賴服務可能會暴露出來多個接口，每個接口就是一個command key

command group，在邏輯上去組織起來一堆command key的調用，統計信息，成功次數，timeout超時次數，失敗次數，可以看到某一個服務整體的一些訪問情況

command group，一般來說，推薦是根據一個服務去劃分出一個線程池，command key默認都是屬于同一個線程池的

command group，對應了一個服務，但是這個服務暴露出來的幾個接口，訪問量很不一樣，差異非常之大

你可能就希望在這個服務command group內部，包含的對應多個接口的command key，做一些細粒度的資源隔離

對同一個服務的不同接口，都使用不同的線程池.

邏輯上來說，多個command key屬于一個command group，在做統計的時候，會放在一起統計

每個command key有自己的線程池，每個接口有自己的線程池，去做資源隔離和限流

但是對于thread pool資源隔離來說，可能是希望能夠拆分的更加一致一些，比如在一個功能模塊內，對不同的請求可以使用不同的thread pool

command group一般來說，可以是對應一個服務，多個command key對應這個服務的多個接口，多個接口的調用共享同一個線程池

如果說你的command key，要用自己的線程池，可以定義自己的threadpool key.

4、coreSize

設置線程池的大小，默認是10

HystrixThreadPoolProperties.Setter()

.withCoreSize(int value)

一般來說，用這個默認的10個線程大小就夠了

5、queueSizeRejectionThreshold

控制queue滿后reject的threshold，因為maxQueueSize不允許熱修改，因此提供這個參數可以熱修改，控制隊列的最大大小

HystrixCommand在提交到線程池之前，其實會先進入一個隊列中，這個隊列滿了之后，才會reject

默認值是5

HystrixThreadPoolProperties.Setter()

.withQueueSizeRejectionThreshold(int value)

6、execution.isolation.semaphore.maxConcurrentRequests

設置使用SEMAPHORE隔離策略的時候，允許訪問的最大并發量，超過這個最大并發量，請求直接被reject

這個并發量的設置，跟線程池大小的設置，應該是類似的，但是基于信號量的話，性能會好很多，而且hystrix框架本身的開銷會小很多

默認值是10，設置的小一些，否則因為信號量是基于調用線程去執行command的，而且不能從timeout中抽離，因此一旦設置的太大，而且有延時發生，可能瞬間導致tomcat本身的線程資源本占滿

HystrixCommandProperties.Setter()

.withExecutionIsolationSemaphoreMaxConcurrentRequests(int value)

創建command，執行這個command，配置這個command對應的group和線程池，以及線程池/信號量的容量和大小

畫圖分析整個8大步驟的流程，然后再對每個步驟進行細致的講解

1、構建一個HystrixCommand或者HystrixObservableCommand

一個HystrixCommand或一個HystrixObservableCommand對象，代表了對某個依賴服務發起的一次請求或者調用

構造的時候，可以在構造函數中傳入任何需要的參數

HystrixCommand主要用于僅僅會返回一個結果的調用

HystrixObservableCommand主要用于可能會返回多條結果的調用

HystrixCommand command = new HystrixCommand(arg1, arg2);

HystrixObservableCommand command = new HystrixObservableCommand(arg1, arg2);

2、調用command的執行方法

執行Command就可以發起一次對依賴服務的調用

要執行Command，需要在4個方法中選擇其中的一個：execute()，queue()，observe()，toObservable()

其中execute()和queue()僅僅對HystrixCommand適用

execute()：調用后直接block住，屬于同步調用，直到依賴服務返回單條結果，或者拋出異常

queue()：返回一個Future，屬于異步調用，后面可以通過Future獲取單條結果

observe()：訂閱一個Observable對象，Observable代表的是依賴服務返回的結果，獲取到一個那個代表結果的Observable對象的拷貝對象

toObservable()：返回一個Observable對象，如果我們訂閱這個對象，就會執行command并且獲取返回結果

K value = command.execute();

Future fValue = command.queue();

Observable ohValue = command.observe();

Observable ocValue = command.toObservable();

execute()實際上會調用queue().get().queue()，接著會調用toObservable().toBlocking().toFuture()

也就是說，無論是哪種執行command的方式，最終都是依賴toObservable()去執行的

3、檢查是否開啟緩存

如果這個command開啟了請求緩存，request cache，而且這個調用的結果在緩存中存在，那么直接從緩存中返回結果

4、檢查是否開啟了短路器

檢查這個command對應的依賴服務是否開啟了短路器

如果斷路器被打開了，那么hystrix就不會執行這個command，而是直接去執行fallback降級機制

5、檢查線程池/隊列/semaphore是否已經滿了

如果command對應的線程池/隊列/semaphore已經滿了，那么也不會執行command，而是直接去調用fallback降級機制

6、執行command

調用HystrixObservableCommand.construct()或HystrixCommand.run()來實際執行這個command

HystrixCommand.run()是返回一個單條結果，或者拋出一個異常

HystrixObservableCommand.construct()是返回一個Observable對象，可以獲取多條結果

如果HystrixCommand.run()或HystrixObservableCommand.construct()的執行，超過了timeout時長的話，那么command所在的線程就會拋出一個TimeoutException

如果timeout了，也會去執行fallback降級機制，而且就不會管run()或construct()返回的值了

這里要注意的一點是，我們是不可能終止掉一個調用嚴重延遲的依賴服務的線程的，只能說給你拋出來一個TimeoutException，但是還是可能會因為嚴重延遲的調用線程占滿整個線程池的

即使這個時候新來的流量都被限流了。。。

如果沒有timeout的話，那么就會拿到一些調用依賴服務獲取到的結果，然后hystrix會做一些logging記錄和metric統計

7、短路健康檢查

Hystrix會將每一個依賴服務的調用成功，失敗，拒絕，超時，等事件，都會發送給circuit breaker斷路器

短路器就會對調用成功/失敗/拒絕/超時等事件的次數進行統計

短路器會根據這些統計次數來決定，是否要進行短路，如果打開了短路器，那么在一段時間內就會直接短路，然后如果在之后第一次檢查發現調用成功了，就關閉斷路器

8、調用fallback降級機制

在以下幾種情況中，hystrix會調用fallback降級機制：run()或construct()拋出一個異常，短路器打開，線程池/隊列/semaphore滿了，command執行超時了

一般在降級機制中，都建議給出一些默認的返回值，比如靜態的一些代碼邏輯，或者從內存中的緩存中提取一些數據，盡量在這里不要再進行網絡請求了

即使在降級中，一定要進行網絡調用，也應該將那個調用放在一個HystrixCommand中，進行隔離

在HystrixCommand中，上線getFallback()方法，可以提供降級機制

在HystirxObservableCommand中，實現一個resumeWithFallback()方法，返回一個Observable對象，可以提供降級結果

如果fallback返回了結果，那么hystrix就會返回這個結果

對于HystrixCommand，會返回一個Observable對象，其中會發返回對應的結果

對于HystrixObservableCommand，會返回一個原始的Observable對象

如果沒有實現fallback，或者是fallback拋出了異常，Hystrix會返回一個Observable，但是不會返回任何數據

不同的command執行方式，其fallback為空或者異常時的返回結果不同

對于execute()，直接拋出異常

對于queue()，返回一個Future，調用get()時拋出異常

對于observe()，返回一個Observable對象，但是調用subscribe()方法訂閱它時，理解拋出調用者的onError方法

對于toObservable()，返回一個Observable對象，但是調用subscribe()方法訂閱它時，理解拋出調用者的onError方法

9、不同的執行方式

execute()，獲取一個Future.get()，然后拿到單個結果

queue()，返回一個Future

observer()，立即訂閱Observable，然后啟動8大執行步驟，返回一個拷貝的Observable，訂閱時理解回調給你結果

toObservable()，返回一個原始的Observable，必須手動訂閱才會去執行8大步驟

總結

以上是生活随笔為你收集整理的hystrix 源码 线程池隔离_基于hystrix的线程池隔离的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。