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 您的支援是對博主最大的鼓勵，感謝您的認真閱讀。 本文版權歸作者所有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連線，否則保留追究法律責任的權利。