RPC框架原理與實現

RPC，全稱 Remote Procedure Call（遠程過程調用），即調用遠程計算機上的服務，就像調用本地服務一樣。那么RPC的原理是什么呢？了解一個技術最好的思路就是尋找一個該類型麻雀雖小五臟俱全的開源項目，不負所期，找到一個輕量級分布式 RPC 框架，本文從這個項目入手來解讀RPC的原理及其實現。

其實說到RPC，大家應該不會陌生才是，以往流行的Web Service就是一種RPC，一般來說RPC 可基於 HTTP 或 TCP 協議，因為Web Service 基於HTTP，所以具有良好的跨平台性，但由於HTTP是應用層協議，相比TCP性能有所損耗。

與本地調用不一樣，遠程調用需要通過網絡層傳輸，因此涉及到的一個問題就是序列化，不同的序列化方式影響調用性能，流行的序列化包括Protobuf、Kryo、Hessian、Jackson、Thrift。

下面，讓我們來一關如何從零開始實現分布式RPC框架。

RPC框架組件

建設一個框架，一個系統，首先要做的就是分析需要哪些組件，他們的關系是什么？

簡單分析下，一個RPC框架需要包括：

• APP ：應用端，調用服務
• Server 服務容器，對外提供服務
• Service Registry 服務注冊表

我們需要將服務部署在分布式環境下的不同節點上，通過服務注冊的方式，讓客戶端來自動發現當前可用的服務，並調用這些服務。這需要一種服務注冊表（Service Registry）的組件，讓它來注冊分布式環境下所有的服務地址（包括：主機名與端口號）。

每台 Server 上可發布多個 Service，這些 Service 共用一個 host 與 port，在分布式環境下會提供 Server 共同對外提供 Service。此外，為防止 Service Registry 出現單點故障，因此需要將其搭建為集群環境。

RPC框架實現

定義服務

首先定義服務接口，接口可以單獨放在一個jar包中

public interface HelloService {

	String hello(String name);

	String hello(Person person);
}

實現接口

然后，增加一種實現

@RpcService(HelloService.class)
public class HelloServiceImpl implements HelloService {

    @Override
    public String hello(String name) {
        return "Hello! " + name;
    }

    @Override
    public String hello(Person person) {
        return "Hello! " + person.getFirstName() + " " + person.getLastName();
    }
}

這里的RpcService注解，定義在服務接口的實現類上，可以讓框架通過這個注解找到服務實現類。

更進一步，如果哪天服務版本升級了，但是歷史服務還有人在使用，怎么辦？解決方案就是服務需要分版本，按版本調用。

@RpcService(value = HelloService.class, version = "sample.hello2")
public class HelloServiceImpl2 implements HelloService {

    @Override
    public String hello(String name) {
        return "你好! " + name;
    }

    @Override
    public String hello(Person person) {
        return "你好! " + person.getFirstName() + " " + person.getLastName();
    }
}

再來看下 RPC 服務注解

/**
 * RPC 服務注解（標注在服務實現類上）
 */
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface RpcService {

    /**
     * 服務接口類
     */
    Class<?> value();

    /**
     * 服務版本號
     */
    String version() default "";
}

服務端實現

Server端主要基於Netty（一個NIO框架）+Spring

回到開頭講的，RPC關鍵點之一就是傳輸序列化，簡單來說就是客戶端調用service時，需要構建一個請求，然后將這個請求序列化傳輸到服務端，服務端完成調用后，再將結果 序列化后返回，簡單畫一下：

定義Request

public class RpcRequest {

    private String requestId;
    private String interfaceName;
    private String serviceVersion;
    private String methodName;
    private Class<?>[] parameterTypes;
    private Object[] parameters;

}

定義RpcResponse

public class RpcResponse {

    private String requestId;
    private Exception exception;
    private Object result;

}

Encoder與Decoder

因為項目基於Netty，所以按Netty那一套搞就行，核心是SerializationUtil，這個根據需要可以采用不同的序列化框架，比如pb。

public class RpcEncoder extends MessageToByteEncoder {

    private Class<?> genericClass;

    public RpcEncoder(Class<?> genericClass) {
        this.genericClass = genericClass;
    }

    @Override
    public void encode(ChannelHandlerContext ctx, Object in, ByteBuf out) throws Exception {
        if (genericClass.isInstance(in)) {
            byte[] data = SerializationUtil.serialize(in);
            out.writeInt(data.length);
            out.writeBytes(data);
        }
    }
}

public class RpcDecoder extends ByteToMessageDecoder {

    private Class<?> genericClass;

    public RpcDecoder(Class<?> genericClass) {
        this.genericClass = genericClass;
    }

    @Override
    public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        if (in.readableBytes() < 4) {
            return;
        }
        in.markReaderIndex();
        int dataLength = in.readInt();
        if (in.readableBytes() < dataLength) {
            in.resetReaderIndex();
            return;
        }
        byte[] data = new byte[dataLength];
        in.readBytes(data);
        out.add(SerializationUtil.deserialize(data, genericClass));
    }
}

掃描服務

服務端采用Spring，並且服務加了RpcService注解，所以服務器啟動的時候掃描一下帶RpcService的就行

下面的代碼實現了將服務找出來，並放到handlerMap里，這樣，調用服務的時候就可以根據服務名稱從Map里找到服務對象，知道了服務對象和服務方法，就能直接調用了。

    private Map<String, Object> handlerMap = new HashMap<>();

    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        // 掃描帶有 RpcService 注解的類並初始化 handlerMap 對象
        Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcService.class);
        if (MapUtils.isNotEmpty(serviceBeanMap)) {
            for (Object serviceBean : serviceBeanMap.values()) {
                RpcService rpcService = serviceBean.getClass().getAnnotation(RpcService.class);
                String serviceName = rpcService.value().getName();
                String serviceVersion = rpcService.version();
                if (StringUtil.isNotEmpty(serviceVersion)) {
                    serviceName += "-" + serviceVersion;
                }
                handlerMap.put(serviceName, serviceBean);
            }
        }
    }

啟動服務器

按照Netty服務器標准代碼，啟動服務，注意Encoder和Decoder

 @Override
    public void afterPropertiesSet() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 創建並初始化 Netty 服務端 Bootstrap 對象
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup);
            bootstrap.channel(NioServerSocketChannel.class);
            bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();
                    pipeline.addLast(new RpcDecoder(RpcRequest.class)); // 解碼 RPC 請求
                    pipeline.addLast(new RpcEncoder(RpcResponse.class)); // 編碼 RPC 響應
                    pipeline.addLast(new RpcServerHandler(handlerMap)); // 處理 RPC 請求
                }
            });
            bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
            // 獲取 RPC 服務器的 IP 地址與端口號
            String[] addressArray = StringUtil.split(serviceAddress, ":");
            String ip = addressArray[0];
            int port = Integer.parseInt(addressArray[1]);
            // 啟動 RPC 服務器
            ChannelFuture future = bootstrap.bind(ip, port).sync();
            // 注冊 RPC 服務地址
            if (serviceRegistry != null) {
                for (String interfaceName : handlerMap.keySet()) {
                    serviceRegistry.register(interfaceName, serviceAddress);
                    LOGGER.debug("register service: {} => {}", interfaceName, serviceAddress);
                }
            }
            LOGGER.debug("server started on port {}", port);
            // 關閉 RPC 服務器
            future.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

處理請求

RpcServerHandler負責處理請求，熟悉Netty的應該知道，繼承SimpleChannelInboundHandler，在channelRead0函數里處理即可，注意，因為pipline里前面已經解碼為RpcRequest對象了，所以在這里可以直接使用。

	public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcServerHandler.class);

    private final Map<String, Object> handlerMap;

    public RpcServerHandler(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx, RpcRequest request) throws Exception {
        // 創建並初始化 RPC 響應對象
        RpcResponse response = new RpcResponse();
        response.setRequestId(request.getRequestId());
        try {
            Object result = handle(request);
            response.setResult(result);
        } catch (Exception e) {
            LOGGER.error("handle result failure", e);
            response.setException(e);
        }
        // 寫入 RPC 響應對象並自動關閉連接
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }
}

框架沒什么說的，核心是怎么handle，無非就是從Reques里獲取到服務名稱和版本號，然后從handlerMap里尋找服務對象，然后調用方法。

已知方法名和Class，可以通過反射進行調用，但是反射性能較低，可以使用cglib里的FastClass來執行invoke，詳情參見說說 cglib 動態代理

    private Object handle(RpcRequest request) throws Exception {
        // 獲取服務對象
        String serviceName = request.getInterfaceName();
        String serviceVersion = request.getServiceVersion();
        if (StringUtil.isNotEmpty(serviceVersion)) {
            serviceName += "-" + serviceVersion;
        }
        Object serviceBean = handlerMap.get(serviceName);
        if (serviceBean == null) {
            throw new RuntimeException(String.format("can not find service bean by key: %s", serviceName));
        }
        // 獲取反射調用所需的參數
        Class<?> serviceClass = serviceBean.getClass();
        String methodName = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameters = request.getParameters();
        // 執行反射調用
//        Method method = serviceClass.getMethod(methodName, parameterTypes);
//        method.setAccessible(true);
//        return method.invoke(serviceBean, parameters);
        // 使用 CGLib 執行反射調用
        FastClass serviceFastClass = FastClass.create(serviceClass);
        FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);
        return serviceFastMethod.invoke(serviceBean, parameters);
    }

服務發現與注冊

在分布式系統里，服務的自動發現與注冊是標配功能，一般來說都是使用集中配置中心，開源屆有Zookeeper、etcd等實現。這里，使用zk作為配置中心。

服務發現與注冊的核心是，服務啟動時，將服務名稱和服務地址寫入到配置中心，客戶端調用的時候，先從集中配置中心讀取所要調用服務的服務器地址，如果有多個，隨機挑選一個（當然隨機的話會存在負載不均衡問題），連接服務器並調用。

個人認為較好的實現方式是，服務層面加一個HA層，客戶端直接調用HA，HA負責負載Service。

回到代碼解讀，這里使用的zookeeper，我們來看怎么實現。

先是定義接口：

public interface ServiceRegistry {

    /**
     * 注冊服務名稱與服務地址
     *
     * @param serviceName    服務名稱
     * @param serviceAddress 服務地址
     */
    void register(String serviceName, String serviceAddress);
}

public interface ServiceDiscovery {

    /**
     * 根據服務名稱查找服務地址
     *
     * @param serviceName 服務名稱
     * @return 服務地址
     */
    String discover(String serviceName);
}

再看談實現，zk有兩種類型的節點，永久節點和臨時節點，這種特性非常適合做服務發現與注冊。
試想：

• 新啟動一台Server，自動注冊到ZK，寫一個臨時節點，客戶端調用的時候就能讀取到這個節點
• 一台Server掛了，臨時節點失效，客戶端調用的時候就讀取不到這個節點，自然不會調用
• 當服務調用量太大，可以新啟動服務，服務小的時候再停掉

不再贅述，看代碼：

public class ZooKeeperServiceRegistry implements ServiceRegistry {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceRegistry.class);

    private final ZkClient zkClient;

    public ZooKeeperServiceRegistry(String zkAddress) {
        // 創建 ZooKeeper 客戶端
        zkClient = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
        LOGGER.debug("connect zookeeper");
    }

    @Override
    public void register(String serviceName, String serviceAddress) {
        // 創建 registry 節點（持久）
        String registryPath = Constant.ZK_REGISTRY_PATH;
        if (!zkClient.exists(registryPath)) {
            zkClient.createPersistent(registryPath);
            LOGGER.debug("create registry node: {}", registryPath);
        }
        // 創建 service 節點（持久）
        String servicePath = registryPath + "/" + serviceName;
        if (!zkClient.exists(servicePath)) {
            zkClient.createPersistent(servicePath);
            LOGGER.debug("create service node: {}", servicePath);
        }
        // 創建 address 節點（臨時）
        String addressPath = servicePath + "/address-";
        String addressNode = zkClient.createEphemeralSequential(addressPath, serviceAddress);
        LOGGER.debug("create address node: {}", addressNode);
    }
}

原理就是創建了一個臨時節點存儲服務地址

再來看服務發現：

public class ZooKeeperServiceDiscovery implements ServiceDiscovery {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceDiscovery.class);

    private String zkAddress;

    public ZooKeeperServiceDiscovery(String zkAddress) {
        this.zkAddress = zkAddress;
    }

    @Override
    public String discover(String name) {
        // 創建 ZooKeeper 客戶端
        ZkClient zkClient = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
        LOGGER.debug("connect zookeeper");
        try {
            // 獲取 service 節點
            String servicePath = Constant.ZK_REGISTRY_PATH + "/" + name;
            if (!zkClient.exists(servicePath)) {
                throw new RuntimeException(String.format("can not find any service node on path: %s", servicePath));
            }
            List<String> addressList = zkClient.getChildren(servicePath);
            if (CollectionUtil.isEmpty(addressList)) {
                throw new RuntimeException(String.format("can not find any address node on path: %s", servicePath));
            }
            // 獲取 address 節點
            String address;
            int size = addressList.size();
            if (size == 1) {
                // 若只有一個地址，則獲取該地址
                address = addressList.get(0);
                LOGGER.debug("get only address node: {}", address);
            } else {
                // 若存在多個地址，則隨機獲取一個地址
                address = addressList.get(ThreadLocalRandom.current().nextInt(size));
                LOGGER.debug("get random address node: {}", address);
            }
            // 獲取 address 節點的值
            String addressPath = servicePath + "/" + address;
            return zkClient.readData(addressPath);
        } finally {
            zkClient.close();
        }
    }
}

客戶端實現

服務代理

可以先查看(http://www.cnblogs.com/xiaoqi/p/java-proxy.html)了解java的動態代理。

使用 Java 提供的動態代理技術實現 RPC 代理（當然也可以使用 CGLib 來實現），具體代碼如下：

public class RpcProxy {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcProxy.class);

    private String serviceAddress;

    private ServiceDiscovery serviceDiscovery;

    public RpcProxy(String serviceAddress) {
        this.serviceAddress = serviceAddress;
    }

    public RpcProxy(ServiceDiscovery serviceDiscovery) {
        this.serviceDiscovery = serviceDiscovery;
    }

    @SuppressWarnings("unchecked")
    public <T> T create(final Class<?> interfaceClass) {
        return create(interfaceClass, "");
    }

    @SuppressWarnings("unchecked")
    public <T> T create(final Class<?> interfaceClass, final String serviceVersion) {
        // 創建動態代理對象
        return (T) Proxy.newProxyInstance(
                interfaceClass.getClassLoader(),
                new Class<?>[]{interfaceClass},
                new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        // 創建 RPC 請求對象並設置請求屬性
                        RpcRequest request = new RpcRequest();
                        request.setRequestId(UUID.randomUUID().toString());
                        request.setInterfaceName(method.getDeclaringClass().getName());
                        request.setServiceVersion(serviceVersion);
                        request.setMethodName(method.getName());
                        request.setParameterTypes(method.getParameterTypes());
                        request.setParameters(args);
                        // 獲取 RPC 服務地址
                        if (serviceDiscovery != null) {
                            String serviceName = interfaceClass.getName();
                            if (StringUtil.isNotEmpty(serviceVersion)) {
                                serviceName += "-" + serviceVersion;
                            }
                            serviceAddress = serviceDiscovery.discover(serviceName);
                            LOGGER.debug("discover service: {} => {}", serviceName, serviceAddress);
                        }
                        if (StringUtil.isEmpty(serviceAddress)) {
                            throw new RuntimeException("server address is empty");
                        }
                        // 從 RPC 服務地址中解析主機名與端口號
                        String[] array = StringUtil.split(serviceAddress, ":");
                        String host = array[0];
                        int port = Integer.parseInt(array[1]);
                        // 創建 RPC 客戶端對象並發送 RPC 請求
                        RpcClient client = new RpcClient(host, port);
                        long time = System.currentTimeMillis();
                        RpcResponse response = client.send(request);
                        LOGGER.debug("time: {}ms", System.currentTimeMillis() - time);
                        if (response == null) {
                            throw new RuntimeException("response is null");
                        }
                        // 返回 RPC 響應結果
                        if (response.hasException()) {
                            throw response.getException();
                        } else {
                            return response.getResult();
                        }
                    }
                }
        );
    }
}

RPC客戶端

使用RpcClient類實現 RPC 客戶端，只需擴展 Netty 提供的SimpleChannelInboundHandler抽象類即可，代碼如下：

public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);

    private final String host;
    private final int port;

    private RpcResponse response;

    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
        this.response = response;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOGGER.error("api caught exception", cause);
        ctx.close();
    }

    public RpcResponse send(RpcRequest request) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            // 創建並初始化 Netty 客戶端 Bootstrap 對象
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();
                    pipeline.addLast(new RpcEncoder(RpcRequest.class)); // 編碼 RPC 請求
                    pipeline.addLast(new RpcDecoder(RpcResponse.class)); // 解碼 RPC 響應
                    pipeline.addLast(RpcClient.this); // 處理 RPC 響應
                }
            });
            bootstrap.option(ChannelOption.TCP_NODELAY, true);
            // 連接 RPC 服務器
            ChannelFuture future = bootstrap.connect(host, port).sync();
            // 寫入 RPC 請求數據並關閉連接
            Channel channel = future.channel();
            channel.writeAndFlush(request).sync();
            channel.closeFuture().sync();
            // 返回 RPC 響應對象
            return response;
        } finally {
            group.shutdownGracefully();
        }
    }
}

服務測試

public class HelloClient {

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new ClassPathXmlApplicationContext("spring.xml");
        RpcProxy rpcProxy = context.getBean(RpcProxy.class);

        HelloService helloService = rpcProxy.create(HelloService.class);
        String result = helloService.hello("World");
        System.out.println(result);

        HelloService helloService2 = rpcProxy.create(HelloService.class, "sample.hello2");
        String result2 = helloService2.hello("世界");
        System.out.println(result2);

        System.exit(0);
    }
}

輸出結果

connect zookeeper
get only address node: address-0000000001
discover service: com.xxx.rpc.sample.api.HelloService => 127.0.0.1:8000
time: 569ms
Hello! World
connect zookeeper
get only address node: address-0000000001
discover service: com.xxx.rpc.sample.api.HelloService-sample.hello2 => 127.0.0.1:8000
time: 36ms
你好! 世界

---

作者：Jadepeng
出處：jqpeng的技術記事本--http://www.cnblogs.com/xiaoqi
您的支持是對博主最大的鼓勵，感謝您的認真閱讀。
本文版權歸作者所有，歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接，否則保留追究法律責任的權利。