简易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

文中代码是依赖上述项目的，若是有不明白的可下载源码

引用

本文中的图取自于网格