从0.5到1写个rpc框架 - 2:远程服务调用(grpc)

acupt

发布于 2019-08-26 11:50:44

8980

发布于 2019-08-26 11:50:44

文章被收录于专栏：一杯82年的JAVA

微服务要实现远程服务调用，除了直接使用如spring coud全家桶中的ribbon、feign模块，也可以试试其他优秀的框架，如谷歌的gRPC，这里基于它实现自己的服务调用模块。

gRPC是Google开源的跨语言远程服务调用(RPC)框架，通信协议用的HTTP/2，数据传输默认用的protocol buffers（一种轻便高效的结构化数据存储格式，想比json更小更快，不过没有可读性）。

需要先掌握grpc的基本用法: gRPC-Java 示例

项目结构

- acuprpc
    + acuprpc-core  //server/client核心处理逻辑
    + acuprpc-protocol-grpc  //基于grpc实现远程调用
    + acuprpc-spring-boot-starter  //server端服务扫描，client端动态代理，服务注册/发现

grpc通信

接口定义

定义服务提供者（server）和服务调用者（client）交流所用的数据结构，client需要告诉server要调用的类名、方法名以及参数（json格式的字符串，在server端再反序列化）。

syntax = "proto3";

option java_multiple_files = true;
option java_package = "com.acupt.acuprpc.protocol.grpc.proto";
option java_outer_classname = "GrpcServiceProto";

package com.acupt.acuprpc.protocol.grpc.proto;

service GrpcService {
    rpc invokeMethod (InvokeRequest) returns (InvokeResponse) {
    }
}

message InvokeRequest {
    string appName = 1;
    string serviceName = 2;
    string methodName = 3;
    repeated string orderedParameter = 4;
    map<string, string> namedParameter = 5;
}

message InvokeResponse {
    int32 code = 1;
    string message = 2;
    string result = 3;
}

grpc-service

这个类负责接收grpc-client发过来的请求，取出请求中的参数，转换成通用的结构，交给core层的RpcServer去执行对应方法，然后将返回值序列化成json返回给grpc-client。

public class GrpcService extends GrpcServiceGrpc.GrpcServiceImplBase {

    private RpcServer rpcServer;

    public GrpcService(RpcServer rpcServer) {
        this.rpcServer = rpcServer;
    }

    @Override
    public void invokeMethod(InvokeRequest request, StreamObserver<InvokeResponse> responseObserver) {
        RpcRequest rpcRequest = new RpcRequest(
                request.getAppName(),
                request.getServiceName(),
                request.getMethodName(),
                request.getOrderedParameterList(),
                request.getNamedParameterMap());
        RpcResponse rpcResponse = rpcServer.execute(rpcRequest);
        InvokeResponse response = InvokeResponse.newBuilder()
                .setCode(rpcResponse.getCode())
                .setMessage(rpcResponse.getMessage())
                .setResult(rpcResponse.getResultString())
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
}

grpc-server

作物服务提供者的具体实现类，只需要实现两个方法：启动服务和关闭服务，其他的交给core层的父类即可。

public class GrpcServer extends RpcServer {

    private Server server;

    public GrpcServer(RpcInstance rpcInstance) {
        super(rpcInstance);
    }

    @SneakyThrows
    @Override
    protected void startRpc() {
        server = ServerBuilder
                .forPort(getRpcInstance().getRpcConf().getPort())
                .addService(new GrpcService(this))
                .build().start();
    }

    @Override
    protected void shutdownRpc() {
        if (server != null) {
            server.shutdown();
        }
    }
}

grpc-client

作为服务调用者，需要把动态代理类传来的请求信息包装成grpc支持的结构，并调用grpc的请求方法，再把远程服务返回的结果返回给代理类。

public class GrpcClient extends RpcClient implements RpcCode {

    private AtomicReference<GrpcServiceGrpc.GrpcServiceFutureStub> stubRef;

    public GrpcClient(NodeInfo nodeInfo) {
        super(nodeInfo);
        this.stubRef = new AtomicReference<>(getStub(nodeInfo));
    }

    @Override
    protected String remoteInvoke(RpcRequest rpcRequest) {
        InvokeRequest.Builder builder = InvokeRequest.newBuilder()
                .setAppName(rpcRequest.getAppName())
                .setServiceName(rpcRequest.getServiceName())
                .setMethodName(rpcRequest.getMethodName());
        // ...
        ListenableFuture<InvokeResponse> future = stubRef.get().invokeMethod(builder.build());
        InvokeResponse response = null;
        //...
        return response.getResult();
    }

    @Override
    @SneakyThrows
    protected NodeInfo reconnectRpc(NodeInfo nodeInfo) {
        //...使用参数中的ip和端口建立新连接，并断开老的连接，可用于重新负载和异常节点重试
    }

    @Override
    @SneakyThrows
    public void shutdownRpc() {
        //...主动断开和服务端的连接
    }

    private GrpcServiceGrpc.GrpcServiceFutureStub getStub(NodeInfo nodeInfo) {
        //...和服务端建立连接，使用参数中的ip和端口
    }

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2019-07-14，如有侵权请联系 cloudcommunity@tencent.com 删除

rpc