grpc雙向流

簡介

grpc是一個高性能、開源和通用的 RPC 框架，面向移動和 HTTP/2 設計。目前提供 C、Java 和 Go 語言版本，分別是：grpc, grpc-java, grpc-go. 其中 C 版本支持 C, C++, Node.js, Python, Ruby, Objective-C, PHP 和 C# 支持.

gRPC 基於 HTTP/2 標準設計，帶來諸如雙向流、流控、頭部壓縮、單 TCP 鏈接上的多複用請求等特。這些特性使得其在移動設備上表現更好，更省電和節省空間佔用。

開發環境配置

首先配置maven引用jar包,導入所有grpc用的包，也可不所有導入，我這裏求方便。

<dependency>
  <groupId>io.grpc</groupId>
  <artifactId>grpc-all</artifactId>
  <version>1.17.1</version>
</dependency>
<dependency>複製代碼

而後引入protobuf文件解析和代碼生成：

<extensions>
   <extension>
      <groupId>kr.motd.maven</groupId>
      <artifactId>os-maven-plugin</artifactId>
      <version>1.5.0.Final</version>
   </extension>
</extensions>
<plugins>
   <plugin>
      <groupId>org.xolstice.maven.plugins</groupId>
      <artifactId>protobuf-maven-plugin</artifactId>
      <version>0.5.1</version>
      <configuration>
         <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
         <pluginId>grpc-java</pluginId>
         <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
      </configuration>
      <executions>
         <execution>
            <goals>
               <goal>compile</goal>
               <goal>compile-custom</goal>
            </goals>
         </execution>
      </executions>
   </plugin>
   <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-enforcer-plugin</artifactId>
      <version>1.4.1</version>
      <executions>
         <execution>
            <id>enforce</id>
            <goals>
               <goal>enforce</goal>
            </goals>
            <configuration>
               <rules>
                  <requireUpperBoundDeps/>
               </rules>
            </configuration>
         </execution>
      </executions>
   </plugin>
</plugins>複製代碼

引入後，在idea中maven projects中的plugins中就會看到protobuf，在這個裏面就能生成所須要的java文件。到「lifecycle」中執行compile或者執行maven命令「mvn compile」，就能生成須要的java文件。

注：定義的protobuf文件要與「java，resources」同級。

入門實例

grpc Client:

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import java.util.concurrent.TimeUnit;


public class HelloWorldClient {

  private final ManagedChannel channel;
  
  private static final GreeterGrpc.GreeterBlockingStub blockingStub;
  
  HelloWorldClient(String host, int port) {
    this.channel =  ManagedChannelBuilder.forAddress(host, port)
            .usePlaintext()
            .build();
    blockingStub = GreeterGrpc.newBlockingStub(channel);
  }

  public void shutdown() throws InterruptedException {
    channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
  }

  public static void main(String[] args) throws Exception {
    HelloWorldClient client = new HelloWorldClient("localhost", 50051);
    try {
      blockingStub.sayHello(HelloRequest.newBuilder().setName("hello").build();) 
    } finally {
      client.shutdown();
    }
  }
}
複製代碼

grpc-server

import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.util.logging.Logger;


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

  private Server server;

  private void start() throws IOException {
    int port = 50051;
    server = ServerBuilder.forPort(port)
        .addService(new GreeterImpl())
        .build()
        .start();
    Runtime.getRuntime().addShutdownHook(new Thread() {
      @Override
      public void run() {
        HelloWorldServer.this.stop();
      }
    });
  }

  private void stop() {
    if (server != null) {
      server.shutdown();
    }
  }

  /**
   * 等待主線程終止，由於GRPC庫使用守護進程線程。
   */
  private void blockUntilShutdown() throws InterruptedException {
    if (server != null) {
      server.awaitTermination();
    }
  }

  public static void main(String[] args) throws IOException, InterruptedException {
    final HelloWorldServer server = new HelloWorldServer();
    server.start();
    server.blockUntilShutdown();
  }

  static class GreeterImpl extends GreeterGrpc.GreeterImplBase {

    @Override
    public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
      HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
      responseObserver.onNext(reply);
      responseObserver.onCompleted();
    }
  }
}
複製代碼

官方入門實例詳見：https://github.com/grpc/grpc-java/tree/master/examples/src/main/java/io/grpc/examples/helloworld

雙向流

server :

Map<String,StreamObserver<Object> > streamObserverMap = new HashMap();
 static class GreeterImpl extends GreeterGrpc.GreeterImplBase {

    @Override
    public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
      HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
          //客戶端調用時，將observer保存起
      streamObserverMap.put("sayHello",responseObserver);
    }
  }    /**
    * 發送消息
    */
    public void sendMsgToclient() {
        //使用時，從保存的streamObserver調用onNext()
        StreamObserver<Object> streamObserver = streamObserverMap.get("sayHello");
        streamObserver.onNext();
        //這裏注意不要調用onCompleted()方法,onCompleted()會將流關閉。固然若是這個流不用了你能夠關閉
//        responseObserver.onCompleted();
    }
複製代碼

Client:

//客戶端要在鏈接上服務端的時候要顯示的調用sendMsg方法,做爲流的註冊
  public static void main(String[] args) throws Exception {
    HelloWorldClient client = new HelloWorldClient("localhost", 50051);
    try {
      //註冊StreamObserver
      blockingStub.sayHello(HelloRequest.newBuilder().setName("").build()) 
    } finally {
      client.shutdown();
    }
  }複製代碼

參考：grpc官方文檔（http://doc.oschina.net/grpc?t=58008）

若有寫的不對地方歡迎在評論指出。