怎么在Java中使用RPC框架過濾器機制

怎么在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();
    }
  }
}

關于怎么在Java中使用RPC框架過濾器機制問題的解答就分享到這里了，希望以上內容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注億速云行業資訊頻道了解更多相關知識。