時間滾動代碼(滾動時間窗口法)
寫在前面，如果喜歡每周分享的干貨內(nèi)容，請留下一個寶貴的贊并且分享給別人，謝謝！
涉及概念
服務等級(service-level)：
核心(core)
重要(important)
普通(normal)
次要(secondary)
非必需(dispensable)
服務隔離
消費者的每個消費的服務之間互相獨立，互不影響，不會因為某個服務的故障或者不可用造成其他服務的故障或者不可用。
隔離策略
線程池隔離： 使用線程池作為隔離的實現(xiàn)方式，每個隔離單元擁有自己單獨的線程池，調(diào)用依賴服務時，申請一個新的線程執(zhí)行真正的調(diào)用邏輯，線程池或者隊列滿了之后，拒絕服務。
信號量隔離： 使用信號量作為隔離的實現(xiàn)方式，每個隔離單元擁有配置了自己的信號量閾值，調(diào)用依賴服務時，在原請求線程中申請新的信號量，如果申請到，繼續(xù)在原線程中執(zhí)行調(diào)用邏輯，信號量超過閾值之后，拒絕服務。
服務限流
按照服務隔離的原則，對每個服務的流量進行限制，不會因為某個或某幾個服務的請求量過大而造成其他服務的不可用
展開全文
服務熔斷
當消費方依賴的某個服務不可用時，動態(tài)的隔絕對該服務的依賴。消費方不再繼續(xù)請求該服務，嘗試使用降級邏輯。當服務恢復可用時，能立即感知并恢復對該服務的依賴。
服務降級
消費方依賴的某個服務不可用（異?；蛘叱瑫r），需要采取的補償性措施。
dubbo與hystrix比較
dubbo的限流，降級方案
消費端通過配置acitves限制消費端調(diào)用的并發(fā)量，在達到最大并發(fā)量之后等待一個timeout時間再重試。
服務端通過配置executes限制服務端接口的線程最大數(shù)量，達到最大數(shù)量之后直接拋出異常。
超時配置，當超時且超過重試次數(shù)之后，拋出異常。消費方實現(xiàn)自己的降級邏輯。
當沒有可用的服務提供者之后，消費者直接短路，消費方實現(xiàn)自己的短路邏輯。
通過注冊中心的URL實現(xiàn)服務運行時參數(shù)的動態(tài)配置。
限流或隔離的粒度是以接口方法為粒度。
dubbo自帶的監(jiān)控不夠強大，需要自己擴展或者使用第三方擴展。
hystrix的限流，降級方案
自定義限流位置。
提供超時時間配置，當超時或者拋出非BadRequestException之后,其他任何錯誤，異?；蛘叱瑫r時，嘗試降級邏輯。
對一段時間內(nèi)的錯誤，超時率進行統(tǒng)計，達到配置的閾值時自動短路，調(diào)用降級邏輯。
服務短路后提供自動恢復機制，快速恢復服務。
通過內(nèi)置的archaius或者第三方配置框架實現(xiàn)服務運行時參數(shù)的動態(tài)配置。
隔離粒度可以自定義，模塊，接口，方法粒度都支持。
提供了基于event-stream的擴展工具和官方的dashboard進行監(jiān)控。但目前官方提供的even-stream是基于servlet的
兩種方案的優(yōu)劣
dubbo同時提供消費端和服務端的限流。hystrix只提供消費端限流。
dubbo的消費端限流的信號量是以服務器為粒度，而hystrix的消費端限流是以整個提供方集群為粒度（更合理）。
dubbo不提供服務容錯降級后的自動短路。hystrix支持自動短路和自動恢復。
dubbo管理平臺中的動態(tài)配置用通知的方式通知消費者，但存在不生效等一些bug。hystrix利用archaius的動態(tài)配置方案從本地或URL中輪詢拉取配置。
dubbo的限流其實是基于信號量的，而hystrix同時支持信號量和線程池的限流。
hystrix與dubbo集成的方案
實現(xiàn)方式
在dubbo的消費端利用dubbo的filter對所有調(diào)用進行攔截擴展代碼如下： DubboHystrixFilter.java
public class DubboHystrixFilter implements Filter {
@Override
public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException {
// 是否啟用hystrix
if (!hystrixIsOpen) {
return invoker.invoke(invocation);
}
String group = invoker.getUrl().getParameter(HystrixConstants.GROUP_KEY);
URL url = invoker.getUrl();
// 未配置groupKey的接口不進行限流
if (StringUtils.isBlank(group)) {
group = invoker.getUrl().getParameter(Constants.ID_KEY);
}
// int serviceLevel = invoker.getUrl().getParameter(HystrixConstants.SERVICE_LEVEL_KEY, ServiceLevelEnum.NORMAL.getLevel());
DubboHystrixCommand command = new DubboHystrixCommand(invoker, invocation, group);
return command.execute();
}
}
public class DubboHystrixFilter implements Filter {
@Override
public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException {
// 是否啟用hystrix
if (!hystrixIsOpen) {
return invoker.invoke(invocation);
}
String group = invoker.getUrl().getParameter(HystrixConstants.GROUP_KEY);
URL url = invoker.getUrl();
// 未配置groupKey的接口不進行限流
if (StringUtils.isBlank(group)) {
group = invoker.getUrl().getParameter(Constants.ID_KEY);
}
// int serviceLevel = invoker.getUrl().getParameter(HystrixConstants.SERVICE_LEVEL_KEY, ServiceLevelEnum.NORMAL.getLevel());
DubboHystrixCommand command = new DubboHystrixCommand(invoker, invocation, group);
return command.execute();
}
}
DubboHystrixCommand.java
public class DubboHystrixCommand extends HystrixCommandResult
{ // 統(tǒng)計一定時間內(nèi)成功請求數(shù)
private static final int STATUSTIME = 20000; // 用于計算百分比的滾動窗口時間長度(毫秒)
private static final int ROLLINGTIME = 60000; private Invoker? invoker; private Invocation invocation; public DubboHystrixCommand(Invoker? invoker, Invocation invocation, String group) { // 使用dubbo配置的優(yōu)先級 method interface application 同等級別 consumer provider
super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(group)) // 組名使用模塊名稱
// 服務等級為NORMAL的隔離粒度為模塊,其他服務等級的隔離粒度為接口
.andCommandKey(HystrixCommandKey.Factory.asKey(group))
.andCommandPropertiesDefaults(
HystrixCommandProperties.Setter()
.withMetricsRollingPercentileWindowInMilliseconds(ROLLINGTIME)
.withMetricsRollingStatisticalWindowInMilliseconds(STATUSTIME) // 使用信號量隔離的方式
.withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)
// 最大并發(fā)量,配置的優(yōu)先級為
.withExecutionIsolationSemaphoreMaxConcurrentRequests(invoker.getUrl().getParameter(
HystrixConstants.CONCURRENCY_KEY, HystrixConstants.DEFAULT_MAX_CONCURRENCY))
.withExecutionTimeoutEnabled(false) // 是否開啟熔斷功能
.withCircuitBreakerEnabled(true)
)); this.invoker = invoker; this.invocation = invocation;
}
@Override protected Result run() throws Exception
{ return invoker.invoke(invocation);
}
@Override protected Result getFallback() { if (executionResult.isResponseSemaphoreRejected()) {
MapString, Object map = new HashMapString, Object(); map.put("resultCode", DHAPCode.COM_FLOW_OVERRUN.getCode()); map.put("resultMsg", DHAPCode.COM_FLOW_OVERRUN.getMsg());
Result result = new RpcResult(map); return result;
}
MapString, Object map = new HashMapString, Object(); map.put("resultCode", DHAPCode.COM_SERVER_ERROR.getCode()); map.put("resultMsg", DHAPCode.COM_SERVER_ERROR.getMsg());
Result result = new RpcResult(map); return result;
}
}
public class DubboHystrixCommand extends HystrixCommandResult
{ // 統(tǒng)計一定時間內(nèi)成功請求數(shù)
private static final int STATUSTIME = 20000; // 用于計算百分比的滾動窗口時間長度(毫秒)
private static final int ROLLINGTIME = 60000; private Invoker? invoker; private Invocation invocation; public DubboHystrixCommand(Invoker? invoker, Invocation invocation, String group) { // 使用dubbo配置的優(yōu)先級 method interface application 同等級別 consumer provider
super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(group)) // 組名使用模塊名稱
// 服務等級為NORMAL的隔離粒度為模塊,其他服務等級的隔離粒度為接口
.andCommandKey(HystrixCommandKey.Factory.asKey(group))
.andCommandPropertiesDefaults(
HystrixCommandProperties.Setter()
.withMetricsRollingPercentileWindowInMilliseconds(ROLLINGTIME)
.withMetricsRollingStatisticalWindowInMilliseconds(STATUSTIME) // 使用信號量隔離的方式
.withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)
// 最大并發(fā)量,配置的優(yōu)先級為
.withExecutionIsolationSemaphoreMaxConcurrentRequests(invoker.getUrl().getParameter(
HystrixConstants.CONCURRENCY_KEY, HystrixConstants.DEFAULT_MAX_CONCURRENCY))
.withExecutionTimeoutEnabled(false) // 是否開啟熔斷功能
.withCircuitBreakerEnabled(true)
)); this.invoker = invoker; this.invocation = invocation;
}
@Override protected Result run() throws Exception
{ return invoker.invoke(invocation);
}
@Override protected Result getFallback() { if (executionResult.isResponseSemaphoreRejected()) {
MapString, Object map = new HashMapString, Object(); map.put("resultCode", DHAPCode.COM_FLOW_OVERRUN.getCode()); map.put("resultMsg", DHAPCode.COM_FLOW_OVERRUN.getMsg());
Result result = new RpcResult(map); return result;
}
MapString, Object map = new HashMapString, Object(); map.put("resultCode", DHAPCode.COM_SERVER_ERROR.getCode()); map.put("resultMsg", DHAPCode.COM_SERVER_ERROR.getMsg());
Result result = new RpcResult(map); return result;
}
}
配置
xmlns:hystrix="http://www.springframework.org/schema/p"dubbo:consumer timeout="60000" check="false" filter="hystrixFilter,consumerLogFilter" hystrix:concurrency="40" /
dubbo:reference id="getUserInfoByIdService" interface="com.cmiot.ums.api.user.GetUserInfoByIdService" hystrix:hgroup="ums" /
xmlns:hystrix="http://www.springframework.org/schema/p"dubbo:consumer timeout="60000" check="false" filter="hystrixFilter,consumerLogFilter" hystrix:concurrency="40" /
dubbo:reference id="getUserInfoByIdService" interface="com.cmiot.ums.api.user.GetUserInfoByIdService" hystrix:hgroup="ums" /
HystrixCommand和HystrixObservableCommand
HystrixObservableCommand只支持異步的調(diào)用方式，HystrixCommand同步異步都支持。
HystrixObservableCommand支持請求合并功能，HystrixCommand不支持。
隔離粒度
對于未配置hystrix:hgroup的消費者不進行限流和熔斷，對于配置了hystrix:hgroup的消費方，默認的最大并發(fā)量為40，隔離粒度根據(jù)配置自定義?？紤]引入服務等級的概念，對于重要的服務默認采用接口級別的隔離粒度，對于非重要框架，每個模塊的每個等級進行隔離，實現(xiàn)對每個服務等級進行動態(tài)調(diào)整。當服務器資源不夠用時，可臨時限制或關(guān)閉非核心服務的功能。
動態(tài)配置
嘗試了使用單獨的配置文件去管理hystrix的配置，但由于我們需要使用dubbo的url中的參數(shù)對服務進行分組，因此如果用獨立的配置文件，配置會比較分散，不易于維護。因此仍然利用dubbo的配置，dubbo的配置也有很多方式，在消費端配置或者在服務端配置，用單一的注冊中心配置還是分模塊的多注冊中心配置。最終我們?nèi)匀粵Q定用在消費端的單一注冊中心配置。理由如下：
服務治理上，服務消費方更清楚服務的使用場景，包括并發(fā)量，重要性等，如何降級，容錯等等。因此，配置在服務消費方比在提供方更合理。
為了解決在服務消費方無法對所使用的服務進行邏輯上的分組，方便分組的統(tǒng)一配置，曾考慮使用多注冊中心的方式，按照每個模塊使用單獨的注冊中配置，但是需要每個服務提供方都去修改注冊中心，改動較大，暫時不采用。
當隔離粒度為模塊時，如果需要變更模塊的配置，目前不太方便，需要對消費方這個分組內(nèi)的所有接口配置，并且可能由于配置時的疏忽造成某個接口與其他接口的配置不同。
hystrix-dashboard
由于hystrix原生的event-stream是基于servlet容器的，應用平臺未使用基于servlet容器的方案，因此對event-stream進行了擴展，方便對接口運行狀況進行實時監(jiān)控統(tǒng)計。 注意hystrix-dashborad在監(jiān)控過程中的請求會被handle住，因此需要配置最大連接數(shù)
隔離策略
線程池策略比較信號量的優(yōu)勢是能夠以非阻塞的方式進行調(diào)用，并且通過對單個接口的壓測顯示性能稍好于信號量。同時，線程池的方式支持緩沖隊列。
信號量比較線程池策略的優(yōu)勢是：相比較動態(tài)調(diào)整大小的開銷比較小，經(jīng)過對單個接口的測試，對CPU的消耗比線程池小。
兩種方式混用
hystrix關(guān)鍵配置
組名(commandGroupKey)
用于統(tǒng)計報表，通知，儀表盤或服務歸屬之類的，使用應用平臺模塊名稱作為commandGroupKey。
命令名(commandKey)
用于監(jiān)控，熔斷器開關(guān)，緩存等作用，也是比較關(guān)鍵的，決定了隔離的粒度，缺省默認使用類名作為key，即根據(jù)當前類名作為隔離粒度，但由于集成采用公用的dubbofilter的機制，所有的名稱都一樣，應用平臺根據(jù)服務等級，大于normal等級的使用接口作為隔離粒度。其他使用模塊級別的隔離粒度。
命令配置(commandProperties)
包含隔離策略配置，線程池大小，信號量大小，超時配置，熔斷功能配置，降級配置，監(jiān)控配置等關(guān)鍵配置信息，每個隔離的單元使用獨立的配置。
super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(group)) // 組名使用模塊名稱
// 服務等級為NORMAL的隔離粒度為模塊,其他服務等級的隔離粒度為接口
.andCommandKey(HystrixCommandKey.Factory.asKey(group))
// 通用配置
.andCommandPropertiesDefaults( HystrixCommandProperties.Setter()
// 是否開啟熔斷
.withCircuitBreakerEnabled(true)
// 觸發(fā)熔斷的錯誤率
.withCircuitBreakerErrorThresholdPercentage(50)
// 強制關(guān)閉熔斷器
.withCircuitBreakerForceClosed(false)
// 強制打開熔斷器
.withCircuitBreakerForceOpen(false)
// 觸發(fā)熔斷器需要的請求量
.withCircuitBreakerRequestVolumeThreshold(20)
// 熔斷器從打開到半開的等待時間
.withCircuitBreakerSleepWindowInMilliseconds(5000)
// 使用信號量隔離的方式 默認:線程池
.withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)
// .withExecutionIsolationStrategy(ExecutionIsolationStrategy.THREAD)
// 信號量閾值 默認:10
.withExecutionIsolationSemaphoreMaxConcurrentRequests(invoker.getUrl().getParameter(HystrixConstants.CONCURRENCY_KEY, HystrixConstants.DEFAULT_MAX_CONCURRENCY))
// // 是否開啟超時時間中斷拋出異常的功能
.withExecutionTimeoutEnabled(false)
// 超時后是否中斷線程
// .withExecutionIsolationThreadInterruptOnTimeout(true)
// // 超時時間 默認:1000// .withExecutionTimeoutInMilliseconds(invoker.getUrl().getParameter(Constants.TIMEOUT_KEY, HystrixConstants.DEFAULT_TIMEOUT_MILLSECOND))
// 是否開啟降級
.withFallbackEnabled(true)
// 信號量隔離時,允許請求降級的最大并發(fā)數(shù)
.withFallbackIsolationSemaphoreMaxConcurrentRequests(10)
// 計算錯誤率的間隔時間
.withMetricsHealthSnapshotIntervalInMilliseconds(500)
// 設置每個bucket內(nèi)執(zhí)行的次數(shù),如果超過這個次數(shù),丟棄最早的，加入最新的
.withMetricsRollingPercentileBucketSize(100)
// 是否開啟監(jiān)控統(tǒng)計功能,如果設置false,任何統(tǒng)計都返回-1
.withMetricsRollingPercentileEnabled(true)
// 用于計算百分比的滾動窗口內(nèi)buckets的個數(shù)
.withMetricsRollingPercentileWindowBuckets(6)
// 用于計算百分比的滾動窗口時間長度
.withMetricsRollingPercentileWindowInMilliseconds(60000)
// 可統(tǒng)計的滾動窗口內(nèi)的buckets數(shù)量,用于熔斷器和指標發(fā)布
.withMetricsRollingStatisticalWindowBuckets(10)
// 可統(tǒng)計的滾動窗口的時間長度,這段時間內(nèi)的執(zhí)行數(shù)據(jù)用于熔斷器和指標發(fā)布
.withMetricsRollingStatisticalWindowInMilliseconds(10000)
// 是否開啟緩存
.withRequestCacheEnabled(true)
// 是否開啟日志
.withRequestLogEnabled(true)
)
// 線程池策略時的配置
// .andThreadPoolPropertiesDefaults(// HystrixThreadPoolProperties.Setter()
// .withCoreSize(10)
// .withKeepAliveTimeMinutes(5)
// .withMaxQueueSize(-1)
// .withMetricsRollingStatisticalWindowBuckets(10)
// .withMetricsRollingStatisticalWindowInMilliseconds(10)
// .withQueueSizeRejectionThreshold(20)
// )
);
super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(group)) // 組名使用模塊名稱
// 服務等級為NORMAL的隔離粒度為模塊,其他服務等級的隔離粒度為接口
.andCommandKey(HystrixCommandKey.Factory.asKey(group))
// 通用配置
.andCommandPropertiesDefaults( HystrixCommandProperties.Setter()
// 是否開啟熔斷
.withCircuitBreakerEnabled(true)
// 觸發(fā)熔斷的錯誤率
.withCircuitBreakerErrorThresholdPercentage(50)
// 強制關(guān)閉熔斷器
.withCircuitBreakerForceClosed(false)
// 強制打開熔斷器
.withCircuitBreakerForceOpen(false)
// 觸發(fā)熔斷器需要的請求量
.withCircuitBreakerRequestVolumeThreshold(20)
// 熔斷器從打開到半開的等待時間
.withCircuitBreakerSleepWindowInMilliseconds(5000)
// 使用信號量隔離的方式 默認:線程池
.withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)
// .withExecutionIsolationStrategy(ExecutionIsolationStrategy.THREAD)
// 信號量閾值 默認:10
.withExecutionIsolationSemaphoreMaxConcurrentRequests(invoker.getUrl().getParameter(HystrixConstants.CONCURRENCY_KEY, HystrixConstants.DEFAULT_MAX_CONCURRENCY))
// // 是否開啟超時時間中斷拋出異常的功能
.withExecutionTimeoutEnabled(false)
// 超時后是否中斷線程
// .withExecutionIsolationThreadInterruptOnTimeout(true)
// // 超時時間 默認:1000// .withExecutionTimeoutInMilliseconds(invoker.getUrl().getParameter(Constants.TIMEOUT_KEY, HystrixConstants.DEFAULT_TIMEOUT_MILLSECOND))
// 是否開啟降級
.withFallbackEnabled(true)
// 信號量隔離時,允許請求降級的最大并發(fā)數(shù)
.withFallbackIsolationSemaphoreMaxConcurrentRequests(10)
// 計算錯誤率的間隔時間
.withMetricsHealthSnapshotIntervalInMilliseconds(500)
// 設置每個bucket內(nèi)執(zhí)行的次數(shù),如果超過這個次數(shù),丟棄最早的，加入最新的
.withMetricsRollingPercentileBucketSize(100)
// 是否開啟監(jiān)控統(tǒng)計功能,如果設置false,任何統(tǒng)計都返回-1
.withMetricsRollingPercentileEnabled(true)
// 用于計算百分比的滾動窗口內(nèi)buckets的個數(shù)
.withMetricsRollingPercentileWindowBuckets(6)
// 用于計算百分比的滾動窗口時間長度
.withMetricsRollingPercentileWindowInMilliseconds(60000)
// 可統(tǒng)計的滾動窗口內(nèi)的buckets數(shù)量,用于熔斷器和指標發(fā)布
.withMetricsRollingStatisticalWindowBuckets(10)
// 可統(tǒng)計的滾動窗口的時間長度,這段時間內(nèi)的執(zhí)行數(shù)據(jù)用于熔斷器和指標發(fā)布
.withMetricsRollingStatisticalWindowInMilliseconds(10000)
// 是否開啟緩存
.withRequestCacheEnabled(true)
// 是否開啟日志
.withRequestLogEnabled(true)
)
// 線程池策略時的配置
// .andThreadPoolPropertiesDefaults(// HystrixThreadPoolProperties.Setter()
// .withCoreSize(10)
// .withKeepAliveTimeMinutes(5)
// .withMaxQueueSize(-1)
// .withMetricsRollingStatisticalWindowBuckets(10)
// .withMetricsRollingStatisticalWindowInMilliseconds(10)
// .withQueueSizeRejectionThreshold(20)
// )
);
springMVC與Hystrix
為每個url或者servlet實現(xiàn)單獨的HystrixCommand，來達到隔離，限流，熔斷，降級的目的
在servlet中的公共filter中實現(xiàn)
在調(diào)用內(nèi)部或者外部服務時用aop的方式實現(xiàn)。
hystrix依賴
hystrix-core中只有三個依賴
archaius-core，用來實現(xiàn)動態(tài)配置，支持擴展第三方工具
rxjava，是hystrix最核心的一部分，是實現(xiàn)異步調(diào)用的核心，基于觀察者模式的擴展，支持基于jvm的語言，例如：scala，groovy等。
HdrHistogram，在hystrix運行時對相關(guān)的metrixs進行收集，支持擴展第三方工具
掃描二維碼推送至手機訪問。
版權(quán)聲明：本文由飛速云SEO網(wǎng)絡優(yōu)化推廣發(fā)布，如需轉(zhuǎn)載請注明出處。
