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

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

namespace java com.acupt.acuprpc.protocol.thrift.proto
service ThriftService{
    InvokeResponse invokeMethod(1:InvokeRequest invokeRequest)
}

struct InvokeRequest{
1: required string appName;
2: required string serviceName;
3: required string methodName;
4: required list<string> orderedParameter;
5: required map<string,string> namedParameter;
}

struct InvokeResponse{
1: required i32 code;
2: optional string message;
3: optional string result;
}

public class ThriftService implements com.acupt.acuprpc.protocol.thrift.proto.ThriftService.Iface {

    private RpcServer rpcServer;

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

    @Override
    public InvokeResponse invokeMethod(InvokeRequest invokeRequest) {
        RpcRequest rpcRequest = new RpcRequest(
                invokeRequest.getAppName(),
                invokeRequest.getServiceName(),
                invokeRequest.getMethodName(),
                invokeRequest.getOrderedParameter(),
                invokeRequest.getNamedParameter());
        RpcResponse rpcResponse = rpcServer.execute(rpcRequest);
        InvokeResponse response = new InvokeResponse();
        response.setCode(rpcResponse.getCode());
        response.setMessage(rpcResponse.getMessage());
        response.setResult(rpcResponse.getResultString());
        return response;
    }
}

public class ThriftServer extends RpcServer {
    private static final int nThreads = 100;
    private TServer server;
    public ThriftServer(RpcInstance rpcInstance) {
        super(rpcInstance);
    }

    @Override
    protected void startRpc() {
        new Thread(() -> {
            TProcessor tprocessor = new com.acupt.acuprpc.protocol.thrift.proto.ThriftService.
                    Processor<com.acupt.acuprpc.protocol.thrift.proto.ThriftService.Iface>(new ThriftService(this));
            TServerTransport serverTransport = null;
            try {
                serverTransport = new TServerSocket(getRpcInstance().getRpcConf().getPort());
            } catch (TTransportException e) {
                throw new RpcException(e);
            }
            TThreadPoolServer.Args tArgs = new TThreadPoolServer.Args(serverTransport);
            tArgs.processor(tprocessor);
            tArgs.executorService(Executors.newFixedThreadPool(nThreads));
            server = new TThreadPoolServer(tArgs);
            server.serve();//阻塞
        }).start();
    }

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

public void send_invokeMethod(InvokeRequest invokeRequest){
    //...
}

public InvokeResponse recv_invokeMethod(){
    //...
}

public InvokeResponse invokeMethod(InvokeRequest invokeRequest) throws org.apache.thrift.TException
{
    send_invokeMethod(invokeRequest);
    return recv_invokeMethod();
}

public class ThriftClient extends RpcClient implements RpcCode {

    private AtomicReference<ThriftService.Client> clientRef;

    public ThriftClient(NodeInfo nodeInfo) {
        super(nodeInfo);
        clientRef = new AtomicReference<>(getClient(nodeInfo));
    }

    //todo client线程不安全，使用连接池管理
    @Override
    @SneakyThrows
    protected synchronized String remoteInvoke(RpcRequest rpcRequest) {
        InvokeRequest request = new InvokeRequest();
        request.setAppName(rpcRequest.getAppName());
        request.setServiceName(rpcRequest.getServiceName());
        request.setMethodName(rpcRequest.getMethodName());
        request.setOrderedParameter(rpcRequest.getOrderedParameter());
        InvokeResponse response = clientRef.get().invokeMethod(request);
        if (response.getCode() != SUCCESS) {
            throw new HttpStatusException(response.getCode(), response.getMessage());
        }
        return response.getResult();
    }

    @Override
    protected NodeInfo reconnectRpc(NodeInfo nodeInfo) {
        //...
    }

    @Override
    public void shutdownRpc() {
        close(clientRef.get());
    }

    private ThriftService.Client getClient(NodeInfo nodeInfo) {
        //...
    }

    private void close(ThriftService.Client client) {
        //...
    }
}