自己实现一个 Mini RPC 框架

自己实现一个 Mini 框架系列之三 —— RPC

查看本系列其他文章

自己实现一个 IoC 框架
自己实现一个 Mini MVC 框架

RPC, 远程过程调用。可以简单地理解为在一台机器上调用另一个地方的代码实现。Java 中比较著名的 RPC 框架有 Dubbo, 公司后端也主要是 Dubbo 支撑起来的。那么自己实现一个 RPC 框架可以吗？

我们希望这样使用（示例代码）

首先是 API 模块，定义 model, 接口。
然后是 provider 模块, 实现接口，然后运行后暴露服务，允许客户端连接过来调用接口的具体实现。
最后是 consumer 模块，引入 api 接口，直接使用接口的方法就可以调用 provider 的实现。

api

IHelloService{
String sayHello(String name);
}

Provider

@Rpc
HelloService{
  @Override
  public String sayHello(String name){
    return "Hello, " + name;
  }
}

Provider{
  psvm{
    Context context = new ClasspathContext("com.youthlin.example"); //扫描 @Rpc注解 后暴露服务
    System.in.read();
    System.exit(0);
  }
}

Consumer

consumer:

package com.youthlin.example.rpc.consumer;

import com.youthlin.example.rpc.api.IHelloService;
import com.youthlin.example.rpc.bean.User;
import com.youthlin.ioc.context.ClasspathContext;
import com.youthlin.ioc.context.Context;
import com.youthlin.rpc.annotation.Rpc;
import com.youthlin.rpc.core.RpcFuture;
import com.youthlin.rpc.util.NetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sun.misc.ProxyGenerator;

import javax.annotation.Resource;
import java.io.File;
import java.io.FileOutputStream;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 创建: youthlin.chen
 * 时间: 2017-11-26 22:38
 */
@Resource
public class Consumer {
    private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
    @Rpc(config = Config.class)
    private IHelloService helloService;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Context context = new ClasspathContext( "com.youthlin.example");
        final Consumer consumer = context.getBean(Consumer.class);
        LOGGER.info("started");
        LOGGER.info("sayHello: {}", consumer.sayHello("World"));
        LOGGER.info("compare: {}", consumer.helloService.compareTo(0));
        LOGGER.info("sayHello: {}", consumer.sayHello("你好"));
        consumer.helloService.clear();
        List<User> userList = consumer.helloService.findAll();
        LOGGER.info("findAll: {}", userList);
        Future<List<User>> future = RpcFuture.get();
        try {
            LOGGER.info("{}", future.get());
        } catch (InterruptedException | ExecutionException e) {
            LOGGER.warn("", e);
        }
        long start = System.nanoTime();
        LOGGER.info("start {}", start);
        consumer.helloService.aLongTimeMethod();
        LOGGER.info("{}", System.nanoTime() - start);
        LOGGER.info("{}", consumer.helloService.toString());
        LOGGER.info("{}", consumer.helloService.getClass());
        LOGGER.info("{}", consumer.helloService.equals(consumer.helloService));

    }

    private String sayHello(String name) {
        return helloService.sayHello(name);
    }

    private void test() {
        for (int i = 0; i < 100; i++) {
            final int count = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException ignore) {
                    }

                    long start = System.currentTimeMillis();
                    helloService.save(new User().setId((long) count).setName("Name" + count));
                    helloService.findAll();
                    Future<List<User>> future = RpcFuture.get();
                    try {
                        List<User> users = future.get(30 * count, TimeUnit.MILLISECONDS);
                        LOGGER.info("No.{} size:{} cost:{} {}", count, users.size(), System.currentTimeMillis() - start, users);
                    } catch (InterruptedException e) {
                        LOGGER.warn("中断异常", e);
                    } catch (ExecutionException e) {
                        LOGGER.warn("调用异常", e);
                    } catch (TimeoutException e) {
                        LOGGER.warn("超时异常", e);

                    }

                }
            }).start();
        }
    }
}

consumer config：

package com.youthlin.example.rpc.consumer;

import com.youthlin.rpc.core.ProxyFactory;
import com.youthlin.rpc.core.SimpleProxyFactory;
import com.youthlin.rpc.core.config.AbstractConsumerConfig;
import com.youthlin.rpc.util.NetUtil;

import java.lang.reflect.Method;

/**
 * 创建: youthlin.chen
 * 时间: 2017-11-27 11:19
 */
public class Config extends AbstractConsumerConfig {
    @Override
    public String host() {
        String host = System.getProperty("provider.host");
        if (host != null && !host.isEmpty()) {
            return host;
        }
        return NetUtil.getLocalAddress().getHostAddress();
    }

    @Override
    public int port() {
        return NetUtil.DEFAULT_PORT;
    }

    @Override
    public Class<? extends ProxyFactory> proxy() {
        return SimpleProxyFactory.class;
    }

    @Override
    public Boolean async(Method method) {
        if (method.getName().equals("findAll")) {
            return true;
        }
        return super.async(method);
    }

    @Override
    public Boolean getConfig(Method method, String key, boolean dft) {
        if (method.getName().equals("aLongTimeMethod") && key.equals(Config.RETURN)) {
            return false;
        }
        return super.getConfig(method, key, dft);
    }
}

大概的流程

provider 端 IoC 容器启动后，扫描有 @Rpc 注解的类, 然后通过注解的 config 属性拿到 ProviderConfig, 其中定义了 Exporter 行为由谁实现, 获取 Exporter 实例，把服务实例暴露出来即可。
consumer 端在 IoC 启动之前，扫描带有 @Rpc 注解的字段，获取 ConsumerConfig, 拿到 ProxyFactory 对该远程接口创建代理，改代理将在容器启动时注入到这个字段，并将代理这个接口的所有方法。

@See Alse: JDK 动态代理的实现

调用时，消费者调用的是创建的代理，代理的实现是把调用信息封装在 Invocation 中通过网络传到服务端。服务端的 Exporter 收到网络请求，解析 Invocation 内容，调用实际 service 把结 set 到 Invocation 中，同样通过网络传回。

基本思路就是这样，但是这个是去年写的，当时比较忙，实现得不太优雅，代码就不贴出来了。有兴趣的同学可以在 github 上看一看：
https://github.com/YouthLin/mini-framework/tree/v1.1.0/mini-rpc
示例代码：https://github.com/YouthLin/examples