阿里面試官：你給我手寫個RPC框架看看-Java面試題-PHP中文網

面試官很喜歡問：

#如何設計一個註冊中心？
如何設計一個訊息佇列？
如何設計一個持久化框架？
如何設計一個RPC框架？
......

今天，就來聊聊「RPC 實作原理」（其他相關如何設計一個XX系列，已發佈到知識星球了）那麼先明確一個問題什麼是RPC 呢？

RPC 是 Remote Procedure Call 的縮寫，即，遠端過程呼叫。

RPC 是一個電腦通訊協定。該協議允許運行於一台計算機的程序調用另一台計算機的子程序，而開發人員無需額外地為這個交互編程。

值得注意是，兩個或多個應用程式都分佈在不同的伺服器上，它們之間的呼叫都像是本機方法呼叫一樣。接下來我們便來分析一次 RPC 呼叫發生了什麼？

RPC 呼叫的基本流程

現在業界比較流行的一些RPC 框架，例如Dubbo 提供的是基於介面的遠端方法調用，即客戶端只需要知道介面的定義即可呼叫遠端服務。在Java 中介面並不能直接呼叫實例方法，必須透過其實作類別物件來完成此操作，這表示客戶端必須為這些介面產生代理對象，對此Java 提供了Proxy、InvocationHandler 產生動態代理程式的支援；生成了代理對象，那麼每個具體的發方法是怎麼呼叫的呢？ JDK 動態代理產生的代理物件呼叫指定方法時實際上會執行 InvocationHandler 中定義的 #invoke 方法，在該方法中完成遠端方法呼叫並取得結果。

拋開客戶端，回過頭來看RPC 是兩台電腦間的調用，實質上是兩台主機間的網路通信，涉及到網路通訊又必然會有序列化、反序列化，編解碼等一些必須要考慮的問題；同時實際上現在大多係統都是叢集部署的，多台主機/容器對外提供相同的服務，如果叢集的節點數量很大的話，那麼管理服務位址也將是一件十分繁瑣的事情，常見的做法是各個服務節點將自己的地址和提供的服務清單註冊到一個註冊中心，由註冊中心來統一管理服務清單；這樣的做法解決了一些問題同時為客戶端增加了一項新的工作——那就是服務發現，通俗來說就是從註冊中心中找到遠端方法對應的服務清單並透過某種策略從中選取一個服務位址來完成網路通訊。

聊了客戶端和註冊中心，另外一個重要的角色自然是服務端，服務端最重要的任務便是提供服務接口的真正實現並在某個端口上監聽網絡請求，監聽到請求後從網路請求中取得對應的參數（例如服務介面、方法、請求參數等），再根據這些參數透過反射的方式呼叫介面的真正實作取得結果並將其寫入對應的回應流中。

綜上所述，一次基本的RPC 呼叫流程大致如下：

基本實作

服務端（生產者）

#服務介面：

在RPC 中，生產者和消費者有一個共同的服務介面API。如下，定義一個 HelloService 介面。

/**
 * @Descrption  服务接口
 ***/
public interface HelloService {
    String sayHello(String somebody);
}

登入後複製

服務實作：

生產者要提供服務介面的實現，建立 HelloServiceImpl 實作類別。

/**
 * @Descrption 服务实现
 ***/
public class HelloServiceImpl implements HelloService {
    @Override
    public String sayHello(String somebody) {
        return "hello " + somebody + "!";
    }
}

登入後複製

服務註冊：

本例使用Spring 來管理bean，採用自訂XML 和解析器的方式來將服務實作類別載入容器（當然也可以採用自訂註解的方式，此處不過多論述）並將服務介面資訊註冊到註冊中心。

首先自訂 XSD：

<xsd:element name="service">
    <xsd:complexType>
        <xsd:complexContent>
            <xsd:extension base="beans:identifiedType">
                <xsd:attribute name="interface" type="xsd:string" use="required"/>
                <xsd:attribute name="timeout" type="xsd:int" use="required"/>
                <xsd:attribute name="serverPort" type="xsd:int" use="required"/>
                <xsd:attribute name="ref" type="xsd:string" use="required"/>
                <xsd:attribute name="weight" type="xsd:int" use="optional"/>
                <xsd:attribute name="workerThreads" type="xsd:int" use="optional"/>
                <xsd:attribute name="appKey" type="xsd:string" use="required"/>
                <xsd:attribute name="groupName" type="xsd:string" use="optional"/>
            </xsd:extension>
        </xsd:complexContent>
    </xsd:complexType>
</xsd:element>

登入後複製

分別指定 Schema 和 XSD，Schema 和對應 Handler 的對應。

Schema：

http\://www.storm.com/schema/storm-service.xsd=META-INF/storm-service.xsd
http\://www.storm.com/schema/storm-reference.xsd=META-INF/storm-reference.xsd

登入後複製

Handler：

http\://www.storm.com/schema/storm-service=com.hsunfkqm.storm.framework.spring.StormServiceNamespaceHandler
http\://www.storm.com/schema/storm-reference=com.hsunfkqm.storm.framework.spring.StormRemoteReferenceNamespaceHandler

登入後複製

將寫好的檔案放入 Classpath 下的 META-INF 目錄下：

在 Spring 配置文件中配置服务类：

<!-- 发布远程服务 -->
 <bean id="helloService" class="com.hsunfkqm.storm.framework.test.HelloServiceImpl"/>
 <storm:service id="helloServiceRegister"
                     interface="com.hsunfkqm.storm.framework.test.HelloService"
                     ref="helloService"
                     groupName="default"
                     weight="2"
                     appKey="ares"
                     workerThreads="100"
                     serverPort="8081"
                     timeout="600"/>

登入後複製

编写对应的 Handler 和 Parser：

StormServiceNamespaceHandler：

import org.springframework.beans.factory.xml.NamespaceHandlerSupport;

/**
 * @author 孙浩
 * @Descrption 服务发布自定义标签
 ***/
public class StormServiceNamespaceHandler extends NamespaceHandlerSupport {
    @Override
    public void init() {
        registerBeanDefinitionParser("service", new ProviderFactoryBeanDefinitionParser());
    }
}

登入後複製

ProviderFactoryBeanDefinitionParser：

protected Class getBeanClass(Element element) {
        return ProviderFactoryBean.class;
    }

    protected void doParse(Element element, BeanDefinitionBuilder bean) {

        try {
            String serviceItf = element.getAttribute("interface");
            String serverPort = element.getAttribute("serverPort");
            String ref = element.getAttribute("ref");
            // ....
            bean.addPropertyValue("serverPort", Integer.parseInt(serverPort));
            bean.addPropertyValue("serviceItf", Class.forName(serviceItf));
            bean.addPropertyReference("serviceObject", ref);
            //...
            if (NumberUtils.isNumber(weight)) {
                bean.addPropertyValue("weight", Integer.parseInt(weight));
            }
            //...
       } catch (Exception e) {
            // ...        
      }
    }

登入後複製

ProviderFactoryBean：

/**
 * @Descrption 服务发布
 ***/
public class ProviderFactoryBean implements FactoryBean, InitializingBean {

    //服务接口
    private Class<?> serviceItf;
    //服务实现
    private Object serviceObject;
    //服务端口
    private String serverPort;
    //服务超时时间
    private long timeout;
    //服务代理对象，暂时没有用到
    private Object serviceProxyObject;
    //服务提供者唯一标识
    private String appKey;
    //服务分组组名
    private String groupName = "default";
    //服务提供者权重，默认为 1 , 范围为 [1-100]
    private int weight = 1;
    //服务端线程数，默认 10 个线程
    private int workerThreads = 10;

    @Override
    public Object getObject() throws Exception {
        return serviceProxyObject;
    }

    @Override
    public Class<?> getObjectType() {
        return serviceItf;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        //启动 Netty 服务端
        NettyServer.singleton().start(Integer.parseInt(serverPort));
        //注册到 zk, 元数据注册中心
        List<ProviderService> providerServiceList = buildProviderServiceInfos();
        IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton();
        registerCenter4Provider.registerProvider(providerServiceList);
    }
}

//================RegisterCenter#registerProvider======================
@Override
public void registerProvider(final List<ProviderService> serviceMetaData) {
    if (CollectionUtils.isEmpty(serviceMetaData)) {
        return;
    }

    //连接 zk, 注册服务
    synchronized (RegisterCenter.class) {
        for (ProviderService provider : serviceMetaData) {
            String serviceItfKey = provider.getServiceItf().getName();

            List<ProviderService> providers = providerServiceMap.get(serviceItfKey);
            if (providers == null) {
                providers = Lists.newArrayList();
            }
            providers.add(provider);
            providerServiceMap.put(serviceItfKey, providers);
        }

        if (zkClient == null) {
            zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
        }

        //创建 ZK 命名空间/当前部署应用 APP 命名空间/
        String APP_KEY = serviceMetaData.get(0).getAppKey();
        String ZK_PATH = ROOT_PATH + "/" + APP_KEY;
        boolean exist = zkClient.exists(ZK_PATH);
        if (!exist) {
            zkClient.createPersistent(ZK_PATH, true);
        }

        for (Map.Entry<String, List<ProviderService>> entry : providerServiceMap.entrySet()) {
            //服务分组
            String groupName = entry.getValue().get(0).getGroupName();
            //创建服务提供者
            String serviceNode = entry.getKey();
            String servicePath = ZK_PATH + "/" + groupName + "/" + serviceNode + "/" + PROVIDER_TYPE;
            exist = zkClient.exists(servicePath);
            if (!exist) {
                zkClient.createPersistent(servicePath, true);
            }

            //创建当前服务器节点
            int serverPort = entry.getValue().get(0).getServerPort();//服务端口
            int weight = entry.getValue().get(0).getWeight();//服务权重
            int workerThreads = entry.getValue().get(0).getWorkerThreads();//服务工作线程
            String localIp = IPHelper.localIp();
            String currentServiceIpNode = servicePath + "/" + localIp + "|" + serverPort + "|" + weight + "|" + workerThreads + "|" + groupName;
            exist = zkClient.exists(currentServiceIpNode);
            if (!exist) {
                //注意，这里创建的是临时节点
                zkClient.createEphemeral(currentServiceIpNode);
            }
            //监听注册服务的变化，同时更新数据到本地缓存
            zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
                @Override
                public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                    if (currentChilds == null) {
                        currentChilds = Lists.newArrayList();
                    }
                    //存活的服务 IP 列表
                    List<String> activityServiceIpList = Lists.newArrayList(Lists.transform(currentChilds, new Function<String, String>() {
                        @Override
                        public String apply(String input) {
                            return StringUtils.split(input, "|")[0];
                        }
                    }));
                    refreshActivityService(activityServiceIpList);
                }
            });

        }
    }
}

登入後複製

至此服务实现类已被载入 Spring 容器中，且服务接口信息也注册到了注册中心。

网络通信：

作为生产者对外提供 RPC 服务，必须有一个网络程序来来监听请求和做出响应。在 Java 领域 Netty 是一款高性能的 NIO 通信框架，很多的框架的通信都是采用 Netty 来实现的，本例中也采用它当做通信服务器。

构建并启动 Netty 服务监听指定端口：

public void start(final int port) {
        synchronized (NettyServer.class) {
            if (bossGroup != null || workerGroup != null) {
                return;
            }

            bossGroup = new NioEventLoopGroup();
            workerGroup = new NioEventLoopGroup();
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap
                    .group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //注册解码器 NettyDecoderHandler
                            ch.pipeline().addLast(new NettyDecoderHandler(StormRequest.class, serializeType));
                            //注册编码器 NettyEncoderHandler
                            ch.pipeline().addLast(new NettyEncoderHandler(serializeType));
                            //注册服务端业务逻辑处理器 NettyServerInvokeHandler
                            ch.pipeline().addLast(new NettyServerInvokeHandler());
                        }
                    });
            try {
                channel = serverBootstrap.bind(port).sync().channel();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

登入後複製

上面的代码中向 Netty 服务的 Pipeline 中添加了编解码和业务处理器，当接收到请求时，经过编解码后，真正处理业务的是业务处理器，即 NettyServerInvokeHandler，该处理器继承自 SimpleChannelInboundHandler，当数据读取完成将触发一个事件，并调用 NettyServerInvokeHandler#channelRead0 方法来处理请求。

@Override
protected void channelRead0(ChannelHandlerContext ctx, StormRequest request) throws Exception {
    if (ctx.channel().isWritable()) {
        //从服务调用对象里获取服务提供者信息
        ProviderService metaDataModel = request.getProviderService();
        long consumeTimeOut = request.getInvokeTimeout();
        final String methodName = request.getInvokedMethodName();

        //根据方法名称定位到具体某一个服务提供者
        String serviceKey = metaDataModel.getServiceItf().getName();
        //获取限流工具类
        int workerThread = metaDataModel.getWorkerThreads();
        Semaphore semaphore = serviceKeySemaphoreMap.get(serviceKey);
        if (semaphore == null) {
            synchronized (serviceKeySemaphoreMap) {
                semaphore = serviceKeySemaphoreMap.get(serviceKey);
                if (semaphore == null) {
                    semaphore = new Semaphore(workerThread);
                    serviceKeySemaphoreMap.put(serviceKey, semaphore);
                }
            }
        }

        //获取注册中心服务
        IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton();
        List<ProviderService> localProviderCaches = registerCenter4Provider.getProviderServiceMap().get(serviceKey);

        Object result = null;
        boolean acquire = false;

        try {
            ProviderService localProviderCache = Collections2.filter(localProviderCaches, new Predicate<ProviderService>() {
                @Override
                public boolean apply(ProviderService input) {
                    return StringUtils.equals(input.getServiceMethod().getName(), methodName);
                }
            }).iterator().next();
            Object serviceObject = localProviderCache.getServiceObject();

            //利用反射发起服务调用
            Method method = localProviderCache.getServiceMethod();
            //利用 semaphore 实现限流
            acquire = semaphore.tryAcquire(consumeTimeOut, TimeUnit.MILLISECONDS);
            if (acquire) {
                result = method.invoke(serviceObject, request.getArgs());
                //System.out.println("---------------"+result);
            }
        } catch (Exception e) {
            System.out.println(JSON.toJSONString(localProviderCaches) + "  " + methodName+" "+e.getMessage());
            result = e;
        } finally {
            if (acquire) {
                semaphore.release();
            }
        }
        //根据服务调用结果组装调用返回对象
        StormResponse response = new StormResponse();
        response.setInvokeTimeout(consumeTimeOut);
        response.setUniqueKey(request.getUniqueKey());
        response.setResult(result);
        //将服务调用返回对象回写到消费端
        ctx.writeAndFlush(response);
    } else {
        logger.error("------------channel closed!---------------");
    }
}

登入後複製

此处还有部分细节如自定义的编解码器等，篇幅所限不在此详述，继承 MessageToByteEncoder 和 ByteToMessageDecoder 覆写对应的 encode 和 decode 方法即可自定义编解码器，使用到的序列化工具如 Hessian/Proto 等可参考对应的官方文档。

请求和响应包装：

为便于封装请求和响应，定义两个 bean 来表示请求和响应。

请求：

/**
 * @author 孙浩
 * @Descrption
 ***/
public class StormRequest implements Serializable {

    private static final long serialVersionUID = -5196465012408804755L;
    //UUID，唯一标识一次返回值
    private String uniqueKey;
    //服务提供者信息
    private ProviderService providerService;
    //调用的方法名称
    private String invokedMethodName;
    //传递参数
    private Object[] args;
    //消费端应用名
    private String appName;
    //消费请求超时时长
    private long invokeTimeout;
    // getter/setter
}

登入後複製

响应：

/**
 * @Descrption
 ***/
public class StormResponse implements Serializable {
    private static final long serialVersionUID = 5785265307118147202L;
    //UUID, 唯一标识一次返回值
    private String uniqueKey;
    //客户端指定的服务超时时间
    private long invokeTimeout;
    //接口调用返回的结果对象
    private Object result;
    //getter/setter
}

登入後複製

客户端（消费者）

客户端（消费者）在 RPC 调用中主要是生成服务接口的代理对象，并从注册中心获取对应的服务列表发起网络请求。

客户端和服务端一样采用 Spring 来管理 bean 解析 XML 配置等不再赘述，重点看下以下几点：

1、通过 JDK 动态代理来生成引入服务接口的代理对象

public Object getProxy() {
    return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{targetInterface}, this);
}

登入後複製

2、从注册中心获取服务列表并依据某种策略选取其中一个服务节点

//服务接口名称
String serviceKey = targetInterface.getName();
//获取某个接口的服务提供者列表
IRegisterCenter4Invoker registerCenter4Consumer = RegisterCenter.singleton();
List<ProviderService> providerServices = registerCenter4Consumer.getServiceMetaDataMap4Consume().get(serviceKey);
//根据软负载策略，从服务提供者列表选取本次调用的服务提供者
ClusterStrategy clusterStrategyService = ClusterEngine.queryClusterStrategy(clusterStrategy);
ProviderService providerService = clusterStrategyService.select(providerServices);

登入後複製

3、通过 Netty 建立连接，发起网络请求

/**
 * @author 孙浩
 * @Descrption Netty 消费端 bean 代理工厂
 ***/
public class RevokerProxyBeanFactory implements InvocationHandler {
    private ExecutorService fixedThreadPool = null;
    //服务接口
    private Class<?> targetInterface;
    //超时时间
    private int consumeTimeout;
    //调用者线程数
    private static int threadWorkerNumber = 10;
    //负载均衡策略
    private String clusterStrategy;

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        ...

        //复制一份服务提供者信息
        ProviderService newProvider = providerService.copy();
        //设置本次调用服务的方法以及接口
        newProvider.setServiceMethod(method);
        newProvider.setServiceItf(targetInterface);

        //声明调用 AresRequest 对象，AresRequest 表示发起一次调用所包含的信息
        final StormRequest request = new StormRequest();
        //设置本次调用的唯一标识
        request.setUniqueKey(UUID.randomUUID().toString() + "-" + Thread.currentThread().getId());
        //设置本次调用的服务提供者信息
        request.setProviderService(newProvider);
        //设置本次调用的方法名称
        request.setInvokedMethodName(method.getName());
        //设置本次调用的方法参数信息
        request.setArgs(args);

        try {
            //构建用来发起调用的线程池
            if (fixedThreadPool == null) {
                synchronized (RevokerProxyBeanFactory.class) {
                    if (null == fixedThreadPool) {
                        fixedThreadPool = Executors.newFixedThreadPool(threadWorkerNumber);
                    }
                }
            }
            //根据服务提供者的 ip,port, 构建 InetSocketAddress 对象，标识服务提供者地址
            String serverIp = request.getProviderService().getServerIp();
            int serverPort = request.getProviderService().getServerPort();
            InetSocketAddress inetSocketAddress = new InetSocketAddress(serverIp, serverPort);
            //提交本次调用信息到线程池 fixedThreadPool, 发起调用
            Future<StormResponse> responseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request));
            //获取调用的返回结果
            StormResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS);
            if (response != null) {
                return response.getResult();
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }
    //  ...
}

登入後複製

Netty 的响应是异步的，为了在方法调用返回前获取到响应结果，需要将异步的结果同步化。

4、Netty 异步返回的结果存入阻塞队列

@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, StormResponse response) throws Exception {
    //将 Netty 异步返回的结果存入阻塞队列，以便调用端同步获取
    RevokerResponseHolder.putResultValue(response);
}

登入後複製

5、请求发出后同步获取结果

//提交本次调用信息到线程池 fixedThreadPool, 发起调用
Future<StormResponse> responseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request));
//获取调用的返回结果
StormResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS);
if (response != null) {
    return response.getResult();
}

//===================================================
//从返回结果容器中获取返回结果，同时设置等待超时时间为 invokeTimeout
long invokeTimeout = request.getInvokeTimeout();
StormResponse response = RevokerResponseHolder.getValue(request.getUniqueKey(), invokeTimeout);

登入後複製

测试

Server：

/**
 * @Descrption
 ***/
public class MainServer {
    public static void main(String[] args) throws Exception {
        //发布服务
        final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("storm-server.xml");
        System.out.println(" 服务发布完成");
    }
}

登入後複製

Client：

public class Client {

    private static final Logger logger = LoggerFactory.getLogger(Client.class);

    public static void main(String[] args) throws Exception {

        final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("storm-client.xml");
        final HelloService helloService = (HelloService) context.getBean("helloService");
        String result = helloService.sayHello("World");
        System.out.println(result);
        for (;;) {

        }
    }
}

登入後複製

结果

生产者：