API网关第三章-分治处理会话流程

一、前言

这一章所做的就是让职责更加分明，Netty服务端只负责接受网络连接调用Session，Session单独拿出来去处理具体的调用逻辑。

二、思路

进行代码重构：

• 划分session和socket两大块
• 把之前NettyServer的部分放到socket部分，把反射调用dubbo的那块留在session部分
• 更加细化出mapperedMethod，这一块是仿照Mybatis的处理，根据打进来的不同类型（GET,PUT,POST,DELETE）方法，switch不同的处理方式。后续应该更细化拆分出Executor。

代码结构

大纲

Mapper模块：

• MapperProxy：延续之前的处理方式，这块是一个Interceptor，用来调用执行后续方法，从直接调用dubbo变成了调用MapperedMethod
• MapperedMethod：switch处理不同的请求，调用GatewaySession中的方法执行具体内容（现有get）。
• MapperProxyFactory：生成IGenericReference的工厂
• MapperRegistry：把uri对应的MapperProxyFactory注册进来
• HttpCommandType：枚举类，对应不同的请求
• HttpStatement：一个HTTP请求的封装类，类似MappedStatement

Session模块：

• GatewaySession： 开启Session的接口，提供get方法，getMapper方法，getConfiguration方法。
• GatewaySessionFactory：Session工厂
• DefaultGatewaySession：实现GatewaySession，主要是实现get，用于调用rpc返回结果。
• DefaultGatewaySessionFactory：创建DefaultGatewaySession的工厂实现类。

Socket模块：

• 这一块就是沿袭之前的netty处理，只是把其中调用服务的逻辑拆分出去，依赖gatewaySession去调用。

三、项目目录

四、代码实现

代码提交到了https://gitcode.net/CreativeAlliance/api-chin/-/tree/dev
代码由于开始重复逻辑过多，只抽出部分来。

MapperedMethod

这块就像做个分流，不同类型的请求往不同地方走。

public Object execute(GatewaySession session, Object args) {
        Object result = null;
        switch (command) {
            case GET:
                result = session.get(uri, args);
                break;
            case POST:
                break;
            case PUT:
                break;
            case DELETE:
                break;
            default:
                throw new RuntimeException("Unknown execution method for: " + command);
        }
        return result;
    }

MapperProxy

相比之前，这里很明显的区别就在于，之前是一个完整的调用dubbo的行为，现在是调用MapperedMethod，然后让其去处理后面session应该执行什么方法。

    @Override
    public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable {
        MapperMethod linkMethod = new MapperMethod(uri, method, gatewaySession.getConfiguration());
        return linkMethod.execute(gatewaySession, args);
    }

DefaultGatewaySession

这里就是真正调用RPC服务的地方，但是这里是写死的，后续应该要完善，去获取参数类型并加入参数。

    @Override
    public Object get(String uri, Object parameter) {

        /* 以下这部分内容，后续拆到执行器中处理 */

        // 配置信息
        HttpStatement httpStatement = configuration.getHttpStatement(uri);
        String application = httpStatement.getApplication();
        String interfaceName = httpStatement.getInterfaceName();

        // 获取基础服务（创建成本较高，内存存放获取）
        ApplicationConfig applicationConfig = configuration.getApplicationConfig(application);
        RegistryConfig registryConfig = configuration.getRegistryConfig(application);
        ReferenceConfig<GenericService> reference = configuration.getReferenceConfig(interfaceName);
        // 构建Dubbo服务
        DubboBootstrap bootstrap = DubboBootstrap.getInstance();
        bootstrap.application(applicationConfig).registry(registryConfig).reference(reference).start();
        // 获取泛化调用服务
        ReferenceConfigCache cache = ReferenceConfigCache.getCache();
        GenericService genericService = cache.get(reference);

        return genericService.$invoke(httpStatement.getMethodName(), new String[]{"java.lang.String"}, new Object[]{"小傅哥"});
    }

五、测试

测试代码

    /**
     * 测试：http://localhost:7397/wg/activity/sayHi
     */
    @Test
    public void test_gateway() throws InterruptedException, ExecutionException {
        // 1. 创建配置信息加载注册
        Configuration configuration = new Configuration();
        HttpStatement httpStatement = new HttpStatement(
                "api-gateway-test",
                "cn.bugstack.gateway.rpc.IActivityBooth",
                "sayHi",
                "/wg/activity/sayHi",
                HttpCommandType.GET);
        configuration.addMapper(httpStatement);

        // 2. 基于配置构建会话工厂
        DefaultGatewaySessionFactory gatewaySessionFactory = new DefaultGatewaySessionFactory(configuration);

        // 3. 创建启动网关网络服务
        GatewaySocketServer server = new GatewaySocketServer(gatewaySessionFactory);

        Future<Channel> future = Executors.newFixedThreadPool(2).submit(server);
        Channel channel = future.get();

        if (null == channel) throw new RuntimeException("netty server start error channel is null");

        while (!channel.isActive()) {
            logger.info("netty server gateway start Ing ...");
            Thread.sleep(500);
        }
        logger.info("netty server gateway start Done! {}", channel.localAddress());

        Thread.sleep(Long.MAX_VALUE);
    }

测试结果

六、总结

还是从测试去分析流程。

• 首先是封装了HttpStatement（对应MappedStatement）
• 并根此进行内部的Mapper注册，便于后续调用
• 开启网关的socket部分的网络服务
• 网络服务方面收到了来自浏览器的请求，会把请求打到session当中，session获得对应的mapper执行相应的方法

写到这里，和Mybatis的映射和会话部分几乎是可以一对一的去理解。