限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

Sentinel 是阿里中間件團(tuán)隊開源的,面向分布式服務(wù)架構(gòu)的輕量級高可用流量控制組件,主要以流量為切入點(diǎn),從流量控制、熔斷降級、系統(tǒng)負(fù)載保護(hù)等多個維度來幫助用戶保護(hù)服務(wù)的穩(wěn)定性。

大家可能會問:Sentinel 和之前常用的熔斷降級庫 Netflix Hystrix 有什么異同呢?Sentinel官網(wǎng)有一個對比的文章,這里摘抄一個總結(jié)的表格,具體的對比可以點(diǎn)此 鏈接 查看。
image

從對比的表格可以看到,Sentinel比Hystrix在功能性上還要強(qiáng)大一些,本文讓我們一起來了解下Sentinel的源碼,揭開Sentinel的神秘面紗。

項目結(jié)構(gòu)

將Sentinel的源碼fork到自己的github庫中,接著把源碼clone到本地,然后開始源碼閱讀之旅吧。

首先我們看一下Sentinel項目的整個結(jié)構(gòu):

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理
  • sentinel-core 核心模塊,限流、降級、系統(tǒng)保護(hù)等都在這里實現(xiàn)
  • sentinel-dashboard 控制臺模塊,可以對連接上的sentinel客戶端實現(xiàn)可視化的管理
  • sentinel-transport 傳輸模塊,提供了基本的監(jiān)控服務(wù)端和客戶端的API接口,以及一些基于不同庫的實現(xiàn)
  • sentinel-extension 擴(kuò)展模塊,主要對DataSource進(jìn)行了部分?jǐn)U展實現(xiàn)
  • sentinel-adapter 適配器模塊,主要實現(xiàn)了對一些常見框架的適配
  • sentinel-demo 樣例模塊,可參考怎么使用sentinel進(jìn)行限流、降級等
  • sentinel-benchmark 基準(zhǔn)測試模塊,對核心代碼的精確性提供基準(zhǔn)測試

運(yùn)行樣例

基本上每個框架都會帶有樣例模塊,有的叫example,有的叫demo,sentinel也不例外。

那我們從sentinel的demo中找一個例子運(yùn)行下看看大致的情況吧,上面說過了sentinel主要的核心功能是做限流、降級和系統(tǒng)保護(hù),那我們就從“限流”開始看sentinel的實現(xiàn)原理吧。

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

可以看到sentinel-demo模塊中有很多不同的樣例,我們找到basic模塊下的flow包,這個包下面就是對應(yīng)的限流的樣例,但是限流也有很多種類型的限流,我們就找根據(jù)qps限流的類看吧,其他的限流方式原理上都大差不差。

public class FlowQpsDemo {

private static final String KEY = "abc";

private static AtomicInteger pass = new AtomicInteger();

private static AtomicInteger block = new AtomicInteger();

private static AtomicInteger total = new AtomicInteger();

private static volatile boolean stop = false;

private static final int threadCount = 32;

private static int seconds = 30;

public static void main(String[] args) throws Exception {

initFlowQpsRule();

tick();

// first make the system run on a very low condition

simulateTraffic();

System.out.println("===== begin to do flow control");

System.out.println("only 20 requests per second can pass");

}

private static void initFlowQpsRule() {

List<FlowRule> rules = new ArrayList<FlowRule>();

FlowRule rule1 = new FlowRule();

rule1.setResource(KEY);

// set limit qps to 20

rule1.setCount(20);

// 設(shè)置限流類型:根據(jù)qps

rule1.setGrade(RuleConstant.FLOW_GRADE_QPS);

rule1.setLimitApp("default");

rules.add(rule1);

// 加載限流的規(guī)則

FlowRuleManager.loadRules(rules);

}

private static void simulateTraffic() {

for (int i = 0; i < threadCount; i++) {

Thread t = new Thread(new RunTask());

t.setName("simulate-traffic-Task");

t.start();

}

}

private static void tick() {

Thread timer = new Thread(new TimerTask());

timer.setName("sentinel-timer-task");

timer.start();

}

static class TimerTask implements Runnable {

@Override

public void run() {

long start = System.currentTimeMillis();

System.out.println("begin to statistic!!!");

long oldTotal = 0;

long oldPass = 0;

long oldBlock = 0;

while (!stop) {

try {

TimeUnit.SECONDS.sleep(1);

} catch (InterruptedException e) {

}

long globalTotal = total.get();

long oneSecondTotal = globalTotal - oldTotal;

oldTotal = globalTotal;

long globalPass = pass.get();

long oneSecondPass = globalPass - oldPass;

oldPass = globalPass;

long globalBlock = block.get();

long oneSecondBlock = globalBlock - oldBlock;

oldBlock = globalBlock;

System.out.println(seconds + " send qps is: " + oneSecondTotal);

System.out.println(TimeUtil.currentTimeMillis() + ", total:" + oneSecondTotal

  • ", pass:" + oneSecondPass

  • ", block:" + oneSecondBlock);

if (seconds-- <= 0) {

stop = true;

}

}

long cost = System.currentTimeMillis() - start;

System.out.println("time cost: " + cost + " ms");

System.out.println("total:" + total.get() + ", pass:" + pass.get()

  • ", block:" + block.get());

System.exit(0);

}

}

static class RunTask implements Runnable {

@Override

public void run() {

while (!stop) {

Entry entry = null;

try {

entry = SphU.entry(KEY);

// token acquired, means pass

pass.addAndGet(1);

} catch (BlockException e1) {

block.incrementAndGet();

} catch (Exception e2) {

// biz exception

} finally {

total.incrementAndGet();

if (entry != null) {

entry.exit();

}

}

Random random2 = new Random();

try {

TimeUnit.MILLISECONDS.sleep(random2.nextInt(50));

} catch (InterruptedException e) {

// ignore

}

}

}

}

}

執(zhí)行上面的代碼后,打印出如下的結(jié)果:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

可以看到,上面的結(jié)果中,pass的數(shù)量和我們的預(yù)期并不相同,我們預(yù)期的是每秒允許pass的請求數(shù)是20個,但是目前有很多pass的請求數(shù)是超過20個的。

原因是,我們這里測試的代碼使用了多線程,注意看 threadCount 的值,一共有32個線程來模擬,而在RunTask的run方法中執(zhí)行資源保護(hù)時,即在 SphU.entry 的內(nèi)部是沒有加鎖的,所以就會導(dǎo)致在高并發(fā)下,pass的數(shù)量會高于20。

可以用下面這個模型來描述下,有一個TimeTicker線程在做統(tǒng)計,每1秒鐘做一次。有N個RunTask線程在模擬請求,被訪問的business code被資源key保護(hù)著,根據(jù)規(guī)則,每秒只允許20個請求通過。

由于pass、block、total等計數(shù)器是全局共享的,而多個RunTask線程在執(zhí)行SphU.entry申請獲取entry時,內(nèi)部沒有鎖保護(hù),所以會存在pass的個數(shù)超過設(shè)定的閾值。

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

那為了證明在單線程下限流的正確性與可靠性,那我們的模型就應(yīng)該變成了這樣:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

那接下來我把 threadCount 的值改為1,只有一個線程來執(zhí)行這個方法,看下具體的限流結(jié)果,執(zhí)行上面的代碼后打印的結(jié)果如下:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

可以看到pass數(shù)基本上維持在20,但是第一次統(tǒng)計的pass值還是超過了20。這又是什么原因?qū)е碌哪兀?/p>

其實仔細(xì)看下Demo中的代碼可以發(fā)現(xiàn),模擬請求是用的一個線程,統(tǒng)計結(jié)果是用的另外一個線程,統(tǒng)計線程每1秒鐘統(tǒng)計一次結(jié)果,這兩個線程之間是有時間上的誤差的。從TimeTicker線程打印出來的時間戳可以看出來,雖然每隔一秒進(jìn)行統(tǒng)計,但是當(dāng)前打印時的時間和上一次的時間還是有誤差的,不完全是1000ms的間隔。

要真正驗證每秒限制20個請求,保證數(shù)據(jù)的精準(zhǔn)性,需要做基準(zhǔn)測試,這個不是本篇文章的重點(diǎn),有興趣的同學(xué)可以去了解下jmh,sentinel中的基準(zhǔn)測試也是通過jmh做的。

深入原理

通過一個簡單的示例程序,我們了解了sentinel可以對請求進(jìn)行限流,除了限流外,還有降級和系統(tǒng)保護(hù)等功能。那現(xiàn)在我們就撥開云霧,深入源碼內(nèi)部去一窺sentinel的實現(xiàn)原理吧。

首先從入口開始: SphU.entry() 。這個方法會去申請一個entry,如果能夠申請成功,則說明沒有被限流,否則會拋出BlockException,表面已經(jīng)被限流了。

從 SphU.entry() 方法往下執(zhí)行會進(jìn)入到 Sph.entry() ,Sph的默認(rèn)實現(xiàn)類是 CtSph ,在CtSph中最終會執(zhí)行到 entry(ResourceWrapperresourceWrapper,intcount,Object...args)throwsBlockException 這個方法。

我們來看一下這個方法的具體實現(xiàn):

public Entry entry(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {

Context context = ContextUtil.getContext();

if (context instanceof NullContext) {

// Init the entry only. No rule checking will occur.

return new CtEntry(resourceWrapper, null, context);

}

if (context == null) {

context = MyContextUtil.myEnter(Constants.CONTEXT_DEFAULT_NAME, "", resourceWrapper.getType());

}

// Global switch is close, no rule checking will do.

if (!Constants.ON) {

return new CtEntry(resourceWrapper, null, context);

}

// 獲取該資源對應(yīng)的SlotChain

ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);

/*

  • Means processor cache size exceeds {@link Constants.MAX_SLOT_CHAIN_SIZE}, so no

  • rule checking will be done.

*/

if (chain == null) {

return new CtEntry(resourceWrapper, null, context);

}

Entry e = new CtEntry(resourceWrapper, chain, context);

try {

// 執(zhí)行Slot的entry方法

chain.entry(context, resourceWrapper, null, count, args);

} catch (BlockException e1) {

e.exit(count, args);

// 拋出BlockExecption

throw e1;

} catch (Throwable e1) {

RecordLog.info("Sentinel unexpected exception", e1);

}

return e;

}

這個方法可以分為以下幾個部分:

  • 1.對參數(shù)和全局配置項做檢測,如果不符合要求就直接返回了一個CtEntry對象,不會再進(jìn)行后面的限流檢測,否則進(jìn)入下面的檢測流程。
  • 2.根據(jù)包裝過的資源對象獲取對應(yīng)的SlotChain
  • 3.執(zhí)行SlotChain的entry方法
  • 3.1.如果SlotChain的entry方法拋出了BlockException,則將該異常繼續(xù)向上拋出
  • 3.2.如果SlotChain的entry方法正常執(zhí)行了,則最后會將該entry對象返回
  • 4.如果上層方法捕獲了BlockException,則說明請求被限流了,否則請求能正常執(zhí)行

其中比較重要的是第2、3兩個步驟,我們來分解一下這兩個步驟。

創(chuàng)建SlotChain

首先看一下lookProcessChain的方法實現(xiàn):

private ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {

ProcessorSlotChain chain = chainMap.get(resourceWrapper);

if (chain == null) {

synchronized (LOCK) {

chain = chainMap.get(resourceWrapper);

if (chain == null)

// Entry size limit.

if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {

return null;

}

// 具體構(gòu)造chain的方法

chain = Env.slotsChainbuilder.build();

Map<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(chainMap.size() + 1);

newMap.putAll(chainMap);

newMap.put(resourceWrapper, chain);

chainMap = newMap;

}

}

}

return chain;

}

該方法使用了一個HashMap做了緩存,key是資源對象。這里加了鎖,并且做了 doublecheck 。具體構(gòu)造chain的方法是通過: Env.slotsChainbuilder.build() 這句代碼創(chuàng)建的。那就進(jìn)入這個方法看看吧。

public ProcessorSlotChain build() {

ProcessorSlotChain chain = new DefaultProcessorSlotChain();

chain.addLast(new NodeSelectorSlot());

chain.addLast(new ClusterBuilderSlot());

chain.addLast(new LogSlot());

chain.addLast(new StatisticSlot());

chain.addLast(new SystemSlot());

chain.addLast(new AuthoritySlot());

chain.addLast(new FlowSlot());

chain.addLast(new DegradeSlot());

return chain;

}

Chain是鏈條的意思,從build的方法可看出,ProcessorSlotChain是一個鏈表,里面添加了很多個Slot。具體的實現(xiàn)需要到DefaultProcessorSlotChain中去看。

public class DefaultProcessorSlotChain extends ProcessorSlotChain {

AbstractLinkedProcessorSlot<?> first = new AbstractLinkedProcessorSlot<Object>() {

@Override

public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, Object... args)

throws Throwable {

super.fireEntry(context, resourceWrapper, t, count, args);

}

@Override

public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {

super.fireExit(context, resourceWrapper, count, args);

}

};

AbstractLinkedProcessorSlot<?> end = first; @Override public void addFirst(AbstractLinkedProcessorSlot<?> protocolProcessor) { protocolProcessor.setNext(first.getNext()); first.setNext(protocolProcessor); if (end == first) { end = protocolProcessor; } } @Override

public void addLast(AbstractLinkedProcessorSlot<?> protocolProcessor)

end.setNext(protocolProcessor);

end = protocolProcessor;

}

}

DefaultProcessorSlotChain中有兩個AbstractLinkedProcessorSlot類型的變量:first和end,這就是鏈表的頭結(jié)點(diǎn)和尾節(jié)點(diǎn)。

創(chuàng)建DefaultProcessorSlotChain對象時,首先創(chuàng)建了首節(jié)點(diǎn),然后把首節(jié)點(diǎn)賦值給了尾節(jié)點(diǎn),可以用下圖表示:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

將第一個節(jié)點(diǎn)添加到鏈表中后,整個鏈表的結(jié)構(gòu)變成了如下圖這樣:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

將所有的節(jié)點(diǎn)都加入到鏈表中后,整個鏈表的結(jié)構(gòu)變成了如下圖所示:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

這樣就將所有的Slot對象添加到了鏈表中去了,每一個Slot都是繼承自AbstractLinkedProcessorSlot。而AbstractLinkedProcessorSlot是一種責(zé)任鏈的設(shè)計,每個對象中都有一個next屬性,指向的是另一個AbstractLinkedProcessorSlot對象。其實責(zé)任鏈模式在很多框架中都有,比如Netty中是通過pipeline來實現(xiàn)的。

知道了SlotChain是如何創(chuàng)建的了,那接下來就要看下是如何執(zhí)行Slot的entry方法的了。

執(zhí)行SlotChain的entry方法

lookProcessChain方法獲得的ProcessorSlotChain的實例是DefaultProcessorSlotChain,那么執(zhí)行chain.entry方法,就會執(zhí)行DefaultProcessorSlotChain的entry方法,而DefaultProcessorSlotChain的entry方法是這樣的:

@Override

public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, Object... args)

throws Throwable {

first.transformEntry(context, resourceWrapper, t, count, args);

}

也就是說,DefaultProcessorSlotChain的entry實際是執(zhí)行的first屬性的transformEntry方法。

而transformEntry方法會執(zhí)行當(dāng)前節(jié)點(diǎn)的entry方法,在DefaultProcessorSlotChain中first節(jié)點(diǎn)重寫了entry方法,具體如下:

@Override

public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, Object... args)

throws Throwable {

super.fireEntry(context, resourceWrapper, t, count, args);

}

first節(jié)點(diǎn)的entry方法,實際又是執(zhí)行的super的fireEntry方法,那繼續(xù)把目光轉(zhuǎn)移到fireEntry方法,具體如下:

@Override

public void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, Object... args)

throws Throwable {

if (next != null) {

next.transformEntry(context, resourceWrapper, obj, count, args);

}

}

從這里可以看到,從fireEntry方法中就開始傳遞執(zhí)行entry了,這里會執(zhí)行當(dāng)前節(jié)點(diǎn)的下一個節(jié)點(diǎn)transformEntry方法,上面已經(jīng)分析過了,transformEntry方法會觸發(fā)當(dāng)前節(jié)點(diǎn)的entry,也就是說fireEntry方法實際是觸發(fā)了下一個節(jié)點(diǎn)的entry方法。具體的流程如下圖所示:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

從圖中可以看出,從最初的調(diào)用Chain的entry()方法,轉(zhuǎn)變成了調(diào)用SlotChain中Slot的entry()方法。從上面的分析可以知道,SlotChain中的第一個Slot節(jié)點(diǎn)是NodeSelectorSlot。

執(zhí)行Slot的entry方法

現(xiàn)在可以把目光轉(zhuǎn)移到SlotChain中的第一個節(jié)點(diǎn)NodeSelectorSlot的entry方法中去了,具體的代碼如下:

@Override

public void entry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, Object... args)

throws Throwable {

DefaultNode node = map.get(context.getName());

if (node == null) {

synchronized (this) {

node = map.get(context.getName());

if (node == null) {

node = Env.nodeBuilder.buildTreeNode(resourceWrapper, null);

HashMap<String, DefaultNode> cacheMap = new HashMap<String, DefaultNode>(map.size());

cacheMap.putAll(map);

cacheMap.put(context.getName(), node);

map = cacheMap;

}

// Build invocation tree

((DefaultNode)context.getLastNode()).addChild(node);

}

}

context.setCurNode(node);

// 由此觸發(fā)下一個節(jié)點(diǎn)的entry方法

fireEntry(context, resourceWrapper, node, count, args);

}

從代碼中可以看到,NodeSelectorSlot節(jié)點(diǎn)做了一些自己的業(yè)務(wù)邏輯處理,具體的大家可以深入源碼繼續(xù)追蹤,這里大概的介紹下每種Slot的功能職責(zé):

  • NodeSelectorSlot 負(fù)責(zé)收集資源的路徑,并將這些資源的調(diào)用路徑,以樹狀結(jié)構(gòu)存儲起來,用于根據(jù)調(diào)用路徑來限流降級;
  • ClusterBuilderSlot 則用于存儲資源的統(tǒng)計信息以及調(diào)用者信息,例如該資源的 RT, QPS, thread count 等等,這些信息將用作為多維度限流,降級的依據(jù);
  • StatistcSlot 則用于記錄,統(tǒng)計不同緯度的 runtime 信息;
  • FlowSlot 則用于根據(jù)預(yù)設(shè)的限流規(guī)則,以及前面 slot 統(tǒng)計的狀態(tài),來進(jìn)行限流;
  • AuthorizationSlot 則根據(jù)黑白名單,來做黑白名單控制;
  • DegradeSlot 則通過統(tǒng)計信息,以及預(yù)設(shè)的規(guī)則,來做熔斷降級;
  • SystemSlot 則通過系統(tǒng)的狀態(tài),例如 load1 等,來控制總的入口流量;

執(zhí)行完業(yè)務(wù)邏輯處理后,調(diào)用了fireEntry()方法,由此觸發(fā)了下一個節(jié)點(diǎn)的entry方法。此時我們就知道了sentinel的責(zé)任鏈就是這樣傳遞的:每個Slot節(jié)點(diǎn)執(zhí)行完自己的業(yè)務(wù)后,會調(diào)用fireEntry來觸發(fā)下一個節(jié)點(diǎn)的entry方法。

所以可以將上面的圖完整了,具體如下:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

至此就通過SlotChain完成了對每個節(jié)點(diǎn)的entry()方法的調(diào)用,每個節(jié)點(diǎn)會根據(jù)創(chuàng)建的規(guī)則,進(jìn)行自己的邏輯處理,當(dāng)統(tǒng)計的結(jié)果達(dá)到設(shè)置的閾值時,就會觸發(fā)限流、降級等事件,具體是拋出BlockException異常。

總結(jié)

sentinel主要是基于7種不同的Slot形成了一個鏈表,每個Slot都各司其職,自己做完分內(nèi)的事之后,會把請求傳遞給下一個Slot,直到在某一個Slot中命中規(guī)則后拋出BlockException而終止。

前三個Slot負(fù)責(zé)做統(tǒng)計,后面的Slot負(fù)責(zé)根據(jù)統(tǒng)計的結(jié)果結(jié)合配置的規(guī)則進(jìn)行具體的控制,是Block該請求還是放行。

控制的類型也有很多可選項:根據(jù)qps、線程數(shù)、冷啟動等等。

然后基于這個核心的方法,衍生出了很多其他的功能:

  • 1、dashboard控制臺,可以可視化的對每個連接過來的sentinel客戶端 (通過發(fā)送heartbeat消息)進(jìn)行控制,dashboard和客戶端之間通過http協(xié)議進(jìn)行通訊。
  • 2、規(guī)則的持久化,通過實現(xiàn)DataSource接口,可以通過不同的方式對配置的規(guī)則進(jìn)行持久化,默認(rèn)規(guī)則是在內(nèi)存中的
  • 3、對主流的框架進(jìn)行適配,包括servlet,dubbo,rRpc等

Dashboard控制臺

sentinel-dashboard是一個單獨(dú)的應(yīng)用,通過spring-boot進(jìn)行啟動,主要提供一個輕量級的控制臺,它提供機(jī)器發(fā)現(xiàn)、單機(jī)資源實時監(jiān)控、集群資源匯總,以及規(guī)則管理的功能。

我們只需要對應(yīng)用進(jìn)行簡單的配置,就可以使用這些功能。

1 啟動控制臺

1.1 下載代碼并編譯控制臺

  • 下載 控制臺 工程
  • 使用以下命令將代碼打包成一個 fat jar: mvn cleanpackage

1.2 啟動

使用如下命令啟動編譯后的控制臺:

$ java -Dserver.port=8080 -Dcsp.sentinel.dashboard.server=localhost:8080 -jar target/sentinel-dashboard.jar

上述命令中我們指定了一個JVM參數(shù), -Dserver.port=8080 用于指定 Spring Boot 啟動端口為 8080。

2 客戶端接入控制臺

控制臺啟動后,客戶端需要按照以下步驟接入到控制臺。

2.1 引入客戶端jar包

通過 pom.xml 引入 jar 包:

<dependency>

<groupId>com.alibaba.csp</groupId>

<artifactId>sentinel-transport-simple-http</artifactId>

<version>x.y.z</version>

</dependency>

2.2 配置啟動參數(shù)

啟動時加入 JVM 參數(shù) -Dcsp.sentinel.dashboard.server=consoleIp:port 指定控制臺地址和端口。若啟動多個應(yīng)用,則需要通過 -Dcsp.sentinel.api.port=xxxx 指定客戶端監(jiān)控 API 的端口(默認(rèn)是 8719)。

除了修改 JVM 參數(shù),也可以通過配置文件取得同樣的效果。更詳細(xì)的信息可以參考 啟動配置項。

2.3 觸發(fā)客戶端初始化

確??蛻舳擞性L問量,Sentinel 會在客戶端首次調(diào)用的時候進(jìn)行初始化,開始向控制臺發(fā)送心跳包。

sentinel-dashboard是一個獨(dú)立的web應(yīng)用,可以接受客戶端的連接,然后與客戶端之間進(jìn)行通訊,他們之間使用http協(xié)議進(jìn)行通訊。他們之間的關(guān)系如下圖所示:

限流降級神器,帶你解讀阿里巴巴開源 Sentinel 實現(xiàn)原理

dashboard

dashboard啟動后會等待客戶端的連接,具體的做法是在 MachineRegistryController 中有一個 receiveHeartBeat 的方法,客戶端發(fā)送心跳消息,就是通過http請求這個方法。

dashboard接收到客戶端的心跳消息后,會把客戶端的傳遞過來的ip、port等信息封裝成一個 MachineInfo對象,然后將該對象通過 MachineDiscovery 接口的 addMachine 方法添加到一個ConcurrentHashMap中保存起來。

這里會有問題,因為客戶端的信息是保存在dashboard的內(nèi)存中的,所以當(dāng)dashboard應(yīng)用重啟后,之前已經(jīng)發(fā)送過來的客戶端信息都會丟失掉。

client

client在啟動時,會通過CommandCenterInitFunc選擇一個,并且只選擇一個CommandCenter進(jìn)行啟動。

啟動之前會通過spi的方式掃描獲取到所有的CommandHandler的實現(xiàn)類,然后將所有的CommandHandler注冊到一個HashMap中去,待后期使用。

PS:考慮一下,為什么CommandHandler不需要做持久化,而是直接保存在內(nèi)存中。

注冊完CommandHandler之后,緊接著就啟動CommandCenter了,目前CommandCenter有兩個實現(xiàn)類:

  • SimpleHttpCommandCenter 通過ServerSocket啟動一個服務(wù)端,接受socket連接
  • NettyHttpCommandCenter 通過Netty啟動一個服務(wù)端,接受channel連接

CommandCenter啟動后,就等待dashboard發(fā)送消息過來了,當(dāng)接收到消息后,會把消息通過具體的CommandHandler進(jìn)行處理,然后將處理的結(jié)果返回給dashboard。

這里需要注意的是,dashboard給client發(fā)送消息是通過異步的httpClient進(jìn)行發(fā)送的,在HttpHelper類中。

但是詭異的是,既然通過異步發(fā)送了,又通過一個CountDownLatch來等待消息的返回,然后獲取結(jié)果,那這樣不就失去了異步的意義的嗎?具體的代碼如下:

private String httpGetContent(String url) { final HttpGet httpGet = new HttpGet(url); final CountDownLatch latch = new CountDownLatch(1);

final AtomicReference<String> reference = new AtomicReference<>();

httpclient.execute(httpGet, new FutureCallback<HttpResponse>() {

@Override

public void completed(final HttpResponse response) {

try {

reference.set(getBody(response));

} catch (Exception e) {

logger.info("httpGetContent " + url + " error:", e);

} finally {

latch.countDown();

}

}

@Override

public void failed(final Exception ex) {

latch.countDown();

logger.info("httpGetContent " + url + " failed:", ex);

}

@Override

public void cancelled() {

latch.countDown();

}

});

try {

latch.await(5, TimeUnit.SECONDS);

} catch (Exception e) {

logger.info("wait http client error:", e);

}

return reference.get();

}

主流框架的適配

sentinel也對一些主流的框架進(jìn)行了適配,使得在使用主流框架時,也可以享受到sentinel的保護(hù)。目前已經(jīng)支持的適配器包括以下這些:

  • Web Servlet
  • Dubbo
  • Spring Boot / Spring Cloud
  • gRPC
  • Apache RocketMQ

其實做適配就是通過那些主流框架的擴(kuò)展點(diǎn),然后在擴(kuò)展點(diǎn)上加入sentinel限流降級的代碼即可。拿Servlet的適配代碼看一下,具體的代碼是:

public class CommonFilter implements Filter {

@Override

public void init(FilterConfig filterConfig) {

}

@Override

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

throws IOException, ServletException

HttpServletRequest sRequest = (HttpServletRequest)request;

Entry entry = null;

try {

// 根據(jù)請求生成的資源

String target = FilterUtil.filterTarget(sRequest);

target = WebCallbackManager.getUrlCleaner().clean(target);

// “申請”該資源

ContextUtil.enter(target);

entry = SphU.entry(target, EntryType.IN);

// 如果能成功“申請”到資源,則說明未被限流

// 則將請求放行

chain.doFilter(request, response);

} catch (BlockException e) {

// 否則如果捕獲了BlockException異常,說明請求被限流了

// 則將請求重定向到一個默認(rèn)的頁面

HttpServletResponse sResponse = (HttpServletResponse)response;

WebCallbackManager.getUrlBlockHandler().blocked(sRequest, sResponse);

} catch (IOException e2) {

// 省略部分代碼

} finally {

if (entry != null) {

entry.exit();

}

ContextUtil.exit();

}

}

@Override

public void destroy() {

}

}

通過Servlet的Filter進(jìn)行擴(kuò)展,實現(xiàn)一個Filter,然后在doFilter方法中對請求進(jìn)行限流控制,如果請求被限流則將請求重定向到一個默認(rèn)頁面,否則將請求放行給下一個Filter。

規(guī)則持久化,動態(tài)化

Sentinel 的理念是開發(fā)者只需要關(guān)注資源的定義,當(dāng)資源定義成功,可以動態(tài)增加各種流控降級規(guī)則。

Sentinel 提供兩種方式修改規(guī)則:

  • 通過 API 直接修改 ( loadRules)
  • 通過 DataSource適配不同數(shù)據(jù)源修改

通過 API 修改比較直觀,可以通過以下三個 API 修改不同的規(guī)則:

FlowRuleManager.loadRules(List<FlowRule> rules); // 修改流控規(guī)則

DegradeRuleManager.loadRules(List<DegradeRule> rules); // 修改降級規(guī)則

SystemRuleManager.loadRules(List<SystemRule> rules); // 修改系統(tǒng)規(guī)則

DataSource 擴(kuò)展

上述 loadRules() 方法只接受內(nèi)存態(tài)的規(guī)則對象,但應(yīng)用重啟后內(nèi)存中的規(guī)則就會丟失,更多的時候規(guī)則最好能夠存儲在文件、數(shù)據(jù)庫或者配置中心中。

DataSource 接口給我們提供了對接任意配置源的能力。相比直接通過 API 修改規(guī)則,實現(xiàn) DataSource 接口是更加可靠的做法。

官方推薦通過控制臺設(shè)置規(guī)則后將規(guī)則推送到統(tǒng)一的規(guī)則中心,用戶只需要實現(xiàn) DataSource 接口,來監(jiān)聽規(guī)則中心的規(guī)則變化,以實時獲取變更的規(guī)則。

DataSource 拓展常見的實現(xiàn)方式有:

  • 拉模式:客戶端主動向某個規(guī)則管理中心定期輪詢拉取規(guī)則,這個規(guī)則中心可以是 SQL、文件,甚至是 VCS 等。這樣做的方式是簡單,缺點(diǎn)是無法及時獲取變更;
  • 推模式:規(guī)則中心統(tǒng)一推送,客戶端通過注冊監(jiān)聽器的方式時刻監(jiān)聽變化,比如使用 Nacos、Zookeeper 等配置中心。這種方式有更好的實時性和一致性保證。

至此,sentinel的基本情況都已經(jīng)分析了,更加詳細(xì)的內(nèi)容,可以繼續(xù)閱讀源碼來研究。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點(diǎn),簡書系信息發(fā)布平臺,僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容