Java RPC框架過濾器機制原理解析
字面義上理解的過濾器類似下圖，從一堆物品中篩選出符合條件的留下，不符合的丟棄。
GOF 職責鏈
GOF中有一種設計模式叫職責鏈，或者叫責任鏈，常規的UML圖如下：
正統的職責鏈是將一個請求發給第一個接收者，接收者判斷是否屬於自己能處理的，如果能處理則執行操作並中止請求下發，流程到此為止。如果不能處理則將請求下發給下一個接收者一直到最後一個接收者。
變體職責鏈
上面提到正統的職責鏈有一個特點：當找到符合條件的執行者後流程中止並不會將請求繼續下發給後續的執行者。這類場景比較適合比如稽核操作，一個報銷單由主管，經理，CTO一級一級的審批不能越權。
解耦合/細分
在實際專案中，往往會遇到非常複雜的業務場景，有可能是需要執行的方法特別多，也有可能是因為需要執行的方法有可能事先不知道，需要在執行時才能判斷。如果將這些邏輯全部寫在一個類或者一個方法中就會出現這樣的問題：
耦合度高，一個方法或者一個類需要關聯所有業務方法以及相關類不易擴充套件，需要執行的業務經常發生變化，如果每次變化都去修改統一的方法或者類，不符合開閉原則，維護成本非常高
所以就有了下圖，一堆需要執行的方法發給第一個接收者，接收者判斷哪些方法是自己可以執行的，有執行的就執行，然後無論是否有可執行的方法在處理完成後都將請求繼續下發給後面的接收者。每個接收者完成自己負責的內容，多個接收者完成了複雜任務的分解。
RPC 過濾器
RPC過濾器與Spring MVC中的Filter作用基本相同，其中很大一個作用就是動態的給客戶端或者是服務端增加切面功能，比如：
許可權控制加密解密訪問日誌限流控制併發控制......
下圖是我在RPC專案中實現過濾器機制的UML示意圖
RpcFilter
定義一個過濾器介面，只包含一個invoke方法。
public interface RpcFilter<T> {
<T> T invoke(RpcInvoker invoker,RpcInvocation invocation);
}
RpcInvoker
這是RPC客戶端以及服務端執行服務端方法或者是將客戶端請求傳送給服務端時需要呼叫的方法介面。
這個角色在Netty中也可以叫Handle，這個介面與上面的RpcFilter有點類似，只是在RPC框架中體現的角色不同而已，具體看UML圖可知道兩者關係。
public interface RpcInvoker {
Object invoke(RpcInvocation invocation);
}
RpcServerInvoker
服務端的一個執行者實現，包含兩個核心功能：
構建職責鏈this.filterMap，是通過註解獲取到的一組過濾器，此處不詳細講因為與本文關係不大RpcInvoker的invoker方法實際呼叫的是RpcFilter的方法，並將自身例項傳遞給RpcFilter，目的是構建職責鏈的關聯關係
public RpcInvoker buildInvokerChain(final RpcInvoker invoker) {
RpcInvoker last = invoker;
List<RpcFilter> filters = Lists.newArrayList(this.filterMap.values());
if (filters.size() > 0) {
for (int i = filters.size() - 1; i >= 0; i --) {
final RpcFilter filter = filters.get(i);
final RpcInvoker next = last;
last = new RpcInvoker() {
@Override
public Object invoke(RpcInvocation invocation) {
return filter.invoke(next,invocation);
}
};
}
}
return last;
}
此處並沒有考慮職責鏈的排序，可以通過過濾器的註解上增加排序數字來解決。目前我寫的過濾器註解中並沒有實現排序功能，可以增加一個order的屬性，然後在需要指定順序的過濾器上增加對應屬性值來支援。
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface ActiveFilter {
String[] group() default {};
String[] value() default {};
}
發請求給職責鏈
服務端在讀取到客戶端的請求後，首先通過構建職責鏈得到RpcInvoker，然後呼叫RpcInvoker的invoke方法將請求下發。
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext,RpcMessage message) {
this.executor.execute(new Runnable() {
@Override
public void run() {
RpcInvoker rpcInvoker=....
RpcResponse response=(RpcResponse) rpcInvoker.invoke......
}
過濾器案例
一般HTTP請求在不同網路角色中處理請求的能力會呈現為一個漏斗型，越上層職責越輕，越往下層職責越重，所對應的就是越上層處理請求量越大，越下層處理請求量越小。比如負載均衡器只負責請求轉發而不負責具體的任務執行，而後端的Service伺服器會執行大量的IO操作或者是消耗cpu的計算任務等，所以這兩者在處理請求的量上往往是數量級的。
當出現大量請求時，為了有效的保護後端服務的穩定性（儘量不出現宕機），除了橫向擴充套件伺服器外還可以通過一些軟體手段緩解後端服務的壓力，這就是通常說的限流，本文因為需要簡單實現一個限制的過濾器，所以直接引用現成的限流演算法：令牌桶。
下面是客戶端請求限流的一個簡單實現，客戶端在給服務端發起請求之前需要獲取令牌，如果獲取到則傳送請求，如果獲取不到一直等待。當然為了防止死鎖，可以呼叫帶超時時間的獲取令牌方法。
@ActiveFilter(group = {ConstantConfig.CONSUMER})
public class AccessLimitFilter implements RpcFilter {
private final static Logger logger = LoggerFactory.getLogger(AccessLimitFilter.class);
@Override
public Object invoke(RpcInvoker invoker,RpcInvocation invocation) {
logger.info("before acquire");
AccessLimitManager.acquire();
Object rpcResponse=invoker.invoke(invocation);
logger.info("after acquire");
return rpcResponse;
}
static class AccessLimitManager{
private final static RateLimiter rateLimiter=RateLimiter.create(2);
public static void acquire(){
rateLimiter.acquire();
}
}
}
實現的比較粗糙，桶的大小是寫死的，應該實現為可配置型，後續抽空完善下。
本文原始碼
https://github.com/jiangmin168168/jim-framework
文中程式碼是依賴上述專案的，如果有不明白的可下載原始碼
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。