JAVA简单实现gRpc服务端和客户端

本文章同步到本人的博客站点 燕归来

远程过程调用（英语：Remote Procedure Call，缩写为 RPC）是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程。如果涉及的软件采用面向对象编程，那么远程过程调用亦可称作远程调用或远程方法调用。简而言之，就是实现不同服务之间的相互调用的这么一个协议，这个不同服务可以是本地服务，也可以是互联网上的远程服务。为了允许不同的客户端均能访问服务器，许多标准化的 RPC 系统应运而生了。其中大部分采用接口描述语言（Interface Description Language，IDL 【对，Android中各个应用之间通讯就是使用的IDL】），方便跨平台的远程过程调用。

今天主要学习Google 开发的一个RPC框架—gRpc这是一个高性能的开源的rpc框架，具有以下特点(翻译的不是很准确)：

• Simple service definition 方便的定义服务
• Works across languages and platforms 跨平台、跨语言
• Start quickly and scale 快速开发和大规模部署
• Bi-directional streaming and integrated auth 双向流设定和认证

下面我们通过一个简单的示例来看下gRpc的使用方法，先把代码附上 GitHub代码地址

需求设定

这里我们假设需要请求服务计算基本的数字运算，客户端发送两个数字，服务端接收到数据数字后计算的到这两个数字的和、差、积。需求很简单，但是不要在客户端计算啊，我们的目的是演示，在客户端计算就没什么意思了....

服务编写

这里我们先说一下，边写的环境信息

• IDEA
• JDK8
• Gralde

注意:build.gradle的配置内容不要随意更改

Proto文件边写

我们需要边写proto文件，文件的格式可以参考Protobuf语言指南——.proto文件语法详解里面讲的很详细，代码如下：

//声明版本
syntax = 'proto3';

//设定一些选项信息
option java_multiple_files = true;
option java_package = "com.tao.example.grpc.basic";
option java_outer_classname = "BasicGprc";
option objc_class_prefix = "HLW";

package basic;

//定义服务
service Grpc {
    
    //定义Rpc,名称为 calculation
    //请求参数类型为  GrpcRequest
    //响应参数类型为  GrpcResponse
    rpc calculation(GrpcRequest) returns(GrpcResponse) {}

}

//在消息定义中，每个字段都有唯一的一个标识符。
//这些标识符是用来在消息的二进制格式中识别各个字段的，一旦开始使用就不能够再改变。


//定义请求参数
message GrpcRequest {
    string num1 = 1;
    string num2 = 2;
}

//定义响应参数
message GrpcResponse {
    string sum = 1;
    string sub = 2;
    string product = 3;
}

完成代码的边写后，在Gradle使用任务去编译这个proto文件 ，任务名称为 generateProto,执行之后在·build/generated/source/proto/main目录下就会生成我们需要的代码，列表如下：

image

服务端代码

服务端代码如下

package com.tao.example;

import com.tao.example.grpc.basic.GrpcGrpc;
import com.tao.example.grpc.basic.GrpcRequest;
import com.tao.example.grpc.basic.GrpcResponse;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.util.Optional;
import java.util.logging.Level;
import java.util.logging.Logger;

public class CalculationService {
  protected static int SERVER_PORT = 8888;

  private static final Logger logger = Logger.getLogger(CalculationService.class.getName());

  private Server server;

  /**
   * 启动服务
   *
   * @param port
   * @throws IOException
   */
  private void start(int port) throws IOException {
    server = ServerBuilder.forPort(port).addService(new BasicCalImpl()).build().start();
    logger.log(Level.INFO, "服务已经启动,监听端口：" + port);
    Runtime.getRuntime()
        .addShutdownHook(
            new Thread(
                () -> {
                  logger.log(Level.WARNING, "监听到JVM停止,正在关闭GRPC服务....");
                  CalculationService.this.stop();
                  logger.log(Level.WARNING, "服务已经停止...");
                }));
  }

  /** 关闭服务 */
  public void stop() {
    Optional.of(server).map(s -> s.shutdown()).orElse(null);
  }

  /**
   * 循环运行服务,封锁停止
   *
   * @throws InterruptedException
   */
  public void blockUnitShutdown() throws InterruptedException {
    if (server != null) {
      server.awaitTermination();
    }
  }

  /**
   * 程序的主运行窗口
   *
   * @param args
   * @throws IOException
   * @throws InterruptedException
   */
  public static void main(String[] args) throws IOException, InterruptedException {
    CalculationService service = new CalculationService();
    service.start(SERVER_PORT);
    service.blockUnitShutdown();
  }

  /** 实现的服务类 */
  static class BasicCalImpl extends GrpcGrpc.GrpcImplBase {

    @Override
    public void calculation(GrpcRequest request, StreamObserver<GrpcResponse> responseObserver) {
      // 获取数据信息
      int num1 = Integer.parseInt(request.getNum1());
      int num2 = Integer.parseInt(request.getNum2());
      // 计算数据
      GrpcResponse response =
          GrpcResponse.newBuilder()
              .setSum(String.valueOf(num1 + num2))
              .setSub(String.valueOf(num1 - num2))
              .setProduct(String.valueOf(num1 * num2))
              .build();
      // 返回数据，完成此次请求
      responseObserver.onNext(response);
      responseObserver.onCompleted();
    }
  }
}

客户端代码

客户端代码和服务端类似，可对比学习。

package com.tao.example;

import com.tao.example.grpc.basic.GrpcGrpc;
import com.tao.example.grpc.basic.GrpcRequest;
import com.tao.example.grpc.basic.GrpcResponse;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.StatusRuntimeException;

import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Pattern;

import static com.tao.example.CalculationService.SERVER_PORT;

public class CalculationClient {
  private static final Logger logger = Logger.getLogger(CalculationClient.class.getName());

  private ManagedChannel managedChannel;

  private GrpcGrpc.GrpcBlockingStub blockingStub;

  public CalculationClient(String host, int port) {
    this(ManagedChannelBuilder.forAddress(host, port).usePlaintext(true));
  }

  public void sendMessage(String num1, String num2) {
    logger.log(Level.INFO, "尝试发送: num1 = " + num1 + ",num2 = " + num2);
    GrpcRequest request = GrpcRequest.newBuilder().setNum1(num1).setNum2(num2).build();
    GrpcResponse response = null;
    try {
      response = blockingStub.calculation(request);
      System.out.println("两数的和 = " + response.getSum());
      System.out.println("两数的差 = " + response.getSub());
      System.out.println("两数的积 = " + response.getProduct());
    } catch (StatusRuntimeException ex) {
      logger.log(Level.WARNING, "发送消息出现异常", ex);
    }
  }

  /**
   * 关闭客户端
   *
   * @throws InterruptedException
   */
  public void shutdown() throws InterruptedException {
    managedChannel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
  }

  CalculationClient(ManagedChannelBuilder<?> channelBuilder) {
    managedChannel = channelBuilder.build();
    blockingStub = GrpcGrpc.newBlockingStub(managedChannel);
  }

  public static void main(String[] args) throws IOException, InterruptedException {
    String host = "127.0.0.1";
    CalculationClient client = new CalculationClient(host, SERVER_PORT);
    Scanner scanner = new Scanner(System.in);
    Pattern pattern = Pattern.compile("^[-\\+]?[\\d]*$");
    System.out.print("请输入Num1:");
    String num1Str = scanner.next();
    if (!pattern.matcher(num1Str).matches()) {
      logger.log(Level.WARNING, "num1不是一个整数,程序无法运行");
    }
    System.out.print("请输入Num2:");
    String num2Str = scanner.next();
    if (!pattern.matcher(num2Str).matches()) {
      logger.log(Level.WARNING, "num2不是一个整数,程序无法运行");
    }
    client.sendMessage(num1Str, num2Str);
  }
}

测试运行

测试运行步骤如下:

• 启动服务器 CalculationService 执行main方法
• 启动测试服务器 CalculationClient 执行main方法
• 在测试服务器控制台窗口输入测试数据，观察结果

启动服务器

image

启动客户端并测试

image

停止服务端

image