安科网

ScalaPB（3）： gRPC streaming

编程语言与高级语言虚拟机杂谈仮

编程语言与高级语言虚拟机杂谈仮

2018-05-11

关注 关注

接着上期讨论的gRPC unary服务我们跟着介绍gRPC streaming，包括: Server-Streaming, Client-Streaming及Bidirectional-Streaming。我们首先在.proto文件里用IDL描述Server-Streaming服务：

/*
 *  responding stream of increment results
 */
service SumOneToMany {
  rpc AddOneToMany(SumRequest) returns (stream SumResponse) {}
}

message SumRequest {
  int32 toAdd = ;
}

message SumResponse {
  int32 currentResult = ;
}

SumOneToMany服务中AddOneToMany函数接受一个SumRequest然后返回stream SumResponse，就这么简单。经过编译后产生了SumOneToManyGrpc.scala文件，在这个文件里提供了有关RPC操作的api。我们看看protoc把IDL描述的服务函数变成了什么样的scala函数：

def addOneToMany(request: SumRequest, responseObserver: StreamObserver[SumResponse]): Unit

调用scala函数addOneToMany需要传入参数SumRequest和StreamObserver[SumResponse]，也就是说用户需要准备这两个入参数。在调用addOneToMany函数时用户事先构建这个StreamObserver传给server，由server把结果通过这个结构传回用户。gRPC是通过StreamObserver类型实例来实现数据streaming的。这个类型的构建例子如下：

val responseObserver = new StreamObserver[SumResponse] {
      def onError(t: Throwable): Unit = println(s"ON_ERROR: $t")
      def onCompleted(): Unit         = println("ON_COMPLETED")
      def onNext(value: SumResponse): Unit =
        println(s"ON_NEXT: Current sum: ${value.currentResult}")
    }

server端通过onNext把结果不断传回给client端，因为这个responseObserver是在client端构建的。下面是SumManyToMany的实现：

class SumOne2ManyService extends SumOneToManyGrpc.SumOneToMany {
    override def addOneToMany(request: SumRequest, responseObserver: StreamObserver[SumResponse]): Unit = {
      val currentSum: AtomicInt = Atomic(0)
      (1 to request.toAdd).map { _ =>
          responseObserver.onNext(SumResponse().withCurrentResult(currentSum.incrementAndGet()))
      }
      Thread.sleep(1000)     //delay and then finish
      responseObserver.onCompleted()
    }
  }

这个addOneToMany服务函数把 1-request.toAdd之间的数字逐个通过responseObserver返还调用方。 在客户端如下调用服务：

// get asyn stub
    val client: SumOneToManyGrpc.SumOneToManyStub = SumOneToManyGrpc.stub(channel)
// prepare stream observer
    val streamObserver = new StreamObserver[SumResponse] {
      override def onError(t: Throwable): Unit = println(s"error: ${t.getMessage}")
      override def onCompleted(): Unit = println("Done incrementing !!!")
      override def onNext(value: SumResponse): Unit = println(s"current value: ${value.currentResult}")
    }
// call service with stream observer
    client.addOneToMany(SumRequest().withToAdd(),streamObserver)

Client-Streaming服务的IDL如下：

/*
 *  responding a result from a request of stream of numbers
 */
service SumManyToOne {
  rpc AddManyToOne(stream SumRequest ) returns (SumResponse) {}
}

传入stream SumRequest, 返回SumResponse。scalaPB自动产生scala代码中的addManyToOne函数款式如下：

def addManyToOne(responseObserver: StreamObserver[SumResponse]): StreamObserver[SumRequest]

调用方提供StreamObserver[SumResponse]用作返回结果，函数返回客方需要的StreamObserver[SumRequest]用以传递request流。注意：虽然在.proto文件中AddManyToOne的返回结果是单个SumResponse，但产生的scala函数则提供了一个StreamObserver[SumResponse]类型，所以需要谨记只能调用一次onNext。下面是这个服务的实现代码：

class Many2OneService extends SumManyToOneGrpc.SumManyToOne {
     val currentSum: AtomicInt = Atomic()
     override def addManyToOne(responseObserver: StreamObserver[SumResponse]): StreamObserver[SumRequest] =
       new StreamObserver[SumRequest] {
         val currentSum: AtomicInt = Atomic()
         override def onError(t: Throwable): Unit = println(s"error: ${t.getMessage}")
         override def onCompleted(): Unit = println("Done summing!")
         override def onNext(value: SumRequest): Unit = {
           //only allow one response
           if (value.toAdd > )
              currentSum.add(value.toAdd)
           else
              responseObserver.onNext(SumResponse(currentSum.addAndGet(value.toAdd)))
         }
       }
   }

客户方调用示范如下：

//pass to server for result
    val respStreamObserver = new StreamObserver[SumResponse] {
      override def onError(t: Throwable): Unit = println(s"error: ${t.getMessage}")
      override def onCompleted(): Unit = println("Done responding!")
      override def onNext(value: SumResponse): Unit =
        println(s"Result: ${value.currentResult}")
    }
    //get async stub
    val client = SumManyToOneGrpc.stub(channel)

    //get request stream observer from server
    val reqStreamObserver = client.addManyToOne(respStreamObserver)

    List(,,,,).map { n =>
      reqStreamObserver.onNext(SumRequest(n))
    }

Bidirectional-Streaming的IDL描述如下：

/*
 * Sums up numbers received from the client and returns the current result after each received request.
 */
service SumInter {
  rpc AddInter(stream SumRequest) returns (stream SumResponse) {}
}

这个service SumInter 描述了stream SumRequest 及 stream SumResponse运算模式。产生的对应scala函数如下：

def addInter(responseObserver: StreamObserver[SumResponse]): StreamObserver[SumRequest]

这个函数的款式与Client-Streaming服务函数是一样的。但是，我们可以通过responseObserver传递多个SumResponse。这个服务的实现代码是这样的：

class Many2ManyService extends SumInterGrpc.SumInter {
    override def addInter(responseObserver: StreamObserver[SumResponse]): StreamObserver[SumRequest] =
      new StreamObserver[SumRequest] {
        val currentSum: AtomicInt = Atomic()
        override def onError(t: Throwable): Unit = println(s"error: ${t.getMessage}")
        override def onCompleted(): Unit = println("Done requesting!")
        override def onNext(value: SumRequest): Unit = {
          responseObserver.onNext(SumResponse(currentSum.addAndGet(value.toAdd)))
        }
      }
  }

我们可以多次调用responseObserver.onNext。客户端源代码如下：

//create stream observer for result stream
    val responseObserver = new StreamObserver[SumResponse] {
      def onError(t: Throwable): Unit = println(s"ON_ERROR: $t")
      def onCompleted(): Unit         = println("ON_COMPLETED")
      def onNext(value: SumResponse): Unit =
        println(s"ON_NEXT: Current sum: ${value.currentResult}")
    }
    //get request container
    val requestObserver = client.addInter(responseObserver)

    scheduler.scheduleWithFixedDelay(0.seconds, 1.seconds) {
      val toBeAdded = Random.nextInt(11)
      println(s"Adding number: $toBeAdded")
      requestObserver.onNext(SumRequest(toBeAdded))
    }

下面是本次示范的源代码：

project/scalapb.sbt

addSbtPlugin("com.thesamet" % "sbt-protoc" % "0.99.18")
libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.7.1"

build.sbt

import scalapb.compiler.Version.scalapbVersion
import scalapb.compiler.Version.grpcJavaVersion

name := "learn-gRPC"

version := "0.1"

scalaVersion := "2.12.6"

libraryDependencies ++= Seq(
  "com.thesamet.scalapb" %% "scalapb-runtime" % scalapbVersion % "protobuf",
  "io.grpc" % "grpc-netty" % grpcJavaVersion,
  "com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapbVersion,
  "io.monix" %% "monix" % "2.3.0"
)

PB.targets in Compile := Seq(
  scalapb.gen() -> (sourceManaged in Compile).value
)

src/main/protobuf/sum.proto

syntax = "proto3";

package learn.grpc.services;

/*
 *  responding stream of increment results
 */
service SumOneToMany {
  rpc AddOneToMany(SumRequest) returns (stream SumResponse) {}
}

/*
 *  responding a result from a request of stream of numbers
 */
service SumManyToOne {
  rpc AddManyToOne(stream SumRequest ) returns (SumResponse) {}
}

/*
 * Sums up numbers received from the client and returns the current result after each received request.
 */
service SumInter {
  rpc AddInter(stream SumRequest) returns (stream SumResponse) {}
}

message SumRequest {
  int32 toAdd = ;
}

message SumResponse {
  int32 currentResult = ;
}

gRPCServer.scala

package learn.grpc.server
import io.grpc.{ServerBuilder,ServerServiceDefinition}

trait gRPCServer {
  def runServer(service: ServerServiceDefinition): Unit = {
    val server = ServerBuilder
      .forPort()
      .addService(service)
      .build
      .start

    // make sure our server is stopped when jvm is shut down
    Runtime.getRuntime.addShutdownHook(new Thread() {
      override def run(): Unit = server.shutdown()
    })

    server.awaitTermination()
  }

}

OneToManyServer.scala

package learn.grpc.sum.one2many.server
import io.grpc.stub.StreamObserver
import learn.grpc.services.sum._
import monix.execution.atomic.{Atomic,AtomicInt}
import learn.grpc.server.gRPCServer

object One2ManyServer extends gRPCServer {

  class SumOne2ManyService extends SumOneToManyGrpc.SumOneToMany {
    override def addOneToMany(request: SumRequest, responseObserver: StreamObserver[SumResponse]): Unit = {
      val currentSum: AtomicInt = Atomic(0)
      (1 to request.toAdd).map { _ =>
          responseObserver.onNext(SumResponse().withCurrentResult(currentSum.incrementAndGet()))
      }
      Thread.sleep(1000)     //delay and then finish
      responseObserver.onCompleted()
    }
  }

  def main(args: Array[String]) = {
    val svc = SumOneToManyGrpc.bindService(new SumOne2ManyService, scala.concurrent.ExecutionContext.global)
    runServer(svc)
  }

}

OneToManyClient.scala

package learn.grpc.sum.one2many.client
import io.grpc.stub.StreamObserver
import learn.grpc.services.sum._

object One2ManyClient {
  def main(args: Array[String]): Unit = {

    //build connection channel
    val channel = io.grpc.ManagedChannelBuilder
      .forAddress("LocalHost",)
      .usePlaintext(true)
      .build()

    // get asyn stub
    val client: SumOneToManyGrpc.SumOneToManyStub = SumOneToManyGrpc.stub(channel)
// prepare stream observer
    val streamObserver = new StreamObserver[SumResponse] {
      override def onError(t: Throwable): Unit = println(s"error: ${t.getMessage}")
      override def onCompleted(): Unit = println("Done incrementing !!!")
      override def onNext(value: SumResponse): Unit = println(s"current value: ${value.currentResult}")
    }
// call service with stream observer
    client.addOneToMany(SumRequest().withToAdd(),streamObserver)

    // wait for async execution
    scala.io.StdIn.readLine()
  }
}

ManyToOneServer.scala

package learn.grpc.sum.many2one.server
import io.grpc.stub.StreamObserver
import learn.grpc.services.sum._
import learn.grpc.server.gRPCServer
import monix.execution.atomic.{Atomic,AtomicInt}

object Many2OneServer extends gRPCServer {
   class Many2OneService extends SumManyToOneGrpc.SumManyToOne {
     val currentSum: AtomicInt = Atomic()
     override def addManyToOne(responseObserver: StreamObserver[SumResponse]): StreamObserver[SumRequest] =
       new StreamObserver[SumRequest] {
         val currentSum: AtomicInt = Atomic()
         override def onError(t: Throwable): Unit = println(s"error: ${t.getMessage}")
         override def onCompleted(): Unit = println("Done summing!")
         override def onNext(value: SumRequest): Unit = {
           //only allow one response
           if (value.toAdd > )
              currentSum.add(value.toAdd)
           else
              responseObserver.onNext(SumResponse(currentSum.addAndGet(value.toAdd)))
         }
       }
   }

   def main(args: Array[String]): Unit = {
     val svc = SumManyToOneGrpc.bindService(new Many2OneService,scala.concurrent.ExecutionContext.global)
     runServer(svc)
   }
}

ManyToOneClient.scala

package learn.grpc.sum.many2one.client
import io.grpc.stub.StreamObserver
import learn.grpc.services.sum._

object Many2OneClient {
  def main(args: Array[String]): Unit = {
    //build channel
    val channel = io.grpc.ManagedChannelBuilder
      .forAddress("LocalHost",)
      .usePlaintext(true)
      .build()
    //pass to server for result
    val respStreamObserver = new StreamObserver[SumResponse] {
      override def onError(t: Throwable): Unit = println(s"error: ${t.getMessage}")
      override def onCompleted(): Unit = println("Done responding!")
      override def onNext(value: SumResponse): Unit =
        println(s"Result: ${value.currentResult}")
    }
    //get async stub
    val client = SumManyToOneGrpc.stub(channel)

    //get request stream observer from server
    val reqStreamObserver = client.addManyToOne(respStreamObserver)

    List(,,,,).map { n =>
      reqStreamObserver.onNext(SumRequest(n))
    }
    scala.io.StdIn.readLine()
  }
}

ManyToManyServer.scala

package learn.grpc.sum.many2many.server
import io.grpc.stub.StreamObserver
import learn.grpc.services.sum._
import learn.grpc.server.gRPCServer
import monix.execution.atomic.{Atomic,AtomicInt}
object Many2ManyServer extends gRPCServer {
  class Many2ManyService extends SumInterGrpc.SumInter {
    override def addInter(responseObserver: StreamObserver[SumResponse]): StreamObserver[SumRequest] =
      new StreamObserver[SumRequest] {
        val currentSum: AtomicInt = Atomic()

        override def onError(t: Throwable): Unit = println(s"error: ${t.getMessage}")

        override def onCompleted(): Unit = println("Done requesting!")

        override def onNext(value: SumRequest): Unit = {
          responseObserver.onNext(SumResponse(currentSum.addAndGet(value.toAdd)))
        }
      }
  }
  def main(args: Array[String]): Unit = {
    val svc = SumInterGrpc.bindService(new Many2ManyService, scala.concurrent.ExecutionContext.global)
    runServer(svc)
  }

}

ManyToManyClient.scala

package learn.grpc.sum.many2many.client
import monix.execution.Scheduler.{global => scheduler}
import learn.grpc.services.sum._

import scala.concurrent.duration._
import scala.util.Random
import io.grpc._
import io.grpc.stub.StreamObserver

object Many2ManyClient {
  def main(args: Array[String]): Unit = {
    val channel = ManagedChannelBuilder.forAddress("localhost", 50051).usePlaintext(true).build
    val client  = SumInterGrpc.stub(channel)
    //create stream observer for result stream
    val responseObserver = new StreamObserver[SumResponse] {
      def onError(t: Throwable): Unit = println(s"ON_ERROR: $t")
      def onCompleted(): Unit         = println("ON_COMPLETED")
      def onNext(value: SumResponse): Unit =
        println(s"ON_NEXT: Current sum: ${value.currentResult}")
    }
    //get request container
    val requestObserver = client.addInter(responseObserver)

    scheduler.scheduleWithFixedDelay(0.seconds, 1.seconds) {
      val toBeAdded = Random.nextInt(11)
      println(s"Adding number: $toBeAdded")
      requestObserver.onNext(SumRequest(toBeAdded))
    }

    scala.io.StdIn.readLine()
  }

}

grpc

编程语言与高级语言虚拟机杂谈仮

编程语言与高级语言虚拟机杂谈仮

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

.Net Core中使用Grpc的方法

　　gRPC 基于如下思想：定义一个服务， 指定其可以被远程调用的方法及其参数和返回类型。gRPC 默认使用protocol buffers作为接口定义语言，来描述服务接口和有效载荷消息结构。如果有需要的话，可以使用其他替代方案。服务端流式 RPC，即客户

darren0zdc 2020-10-16

java 通过 Grpc 与c 通信

由于项目是java 和 c 开发的并且各自独立部署，所以涉及业务需要进行java 与c 通信；根据自测grpc 客户端发现，grpc 支持自动重连、自动判断服务是否上线；

ltstud 2020-07-24

思考gRPC ：为什么是HTTP/2

§。§

岁月如歌 2020-07-21

grpc 安装

安装请参考官方资料-快速开始，安装gRPC需要安装Go、protocol buffer、protoc-gen-go、gRPC等。windows下直接下载编译好的包，然后将bin目录加到环境变量。这是因为grpc的版本和protoc-gen-go版本不一致造

Erick 2020-07-18

golang使用grpc

我的网络通信概念还停留在比较基础的地方, 直觉来说就是发送一个get/post请求, 设定一定的参数格式, 这样对方再解析你的数据, 基于此完成通讯.而rpc则不使用这种显式的通讯方式, 通过proto与生成grpc文件, 里面提供方法, 供client与

wujingsheng0 2020-06-16

Asp.net core 通过grpc调用python

简单来说就是远程过程调用。举个例子，比如在C#代码中直接调用python 中写的方法，或者一个C#进程调用另一个C#进程中的方法，或者其他语言之间的互相调用。具体书写方法可百度。syntax = "proto3"; //表示用的是pro

xmwang0 2020-06-14

【gRPC】 在.Net core中使用gRPC

gRPC是一个高性能的通信协议，它基于HTTP/2和protocol buffers。它是微服务之间进行同步通信的主要选择。与之相对的，就是其他协议，如AMQP的异步通信队列或者发布/订阅模式。RPC框架实际是提供了一套机制，使得应用程序之间可以进行通信，

极品小肥羊 2020-05-31

微服务开发手册之GRPC

> GRPC是一个高性能、通用的开源RPC框架，基于HTTP/2协议标准和Protobuf序列化协议开发，支持众多的开发语言。gRPC使用protocol buffers作为接口描述语言以及底层的信息交换格式，一般情况下推荐使用 proto3因为其能

小小书童 2020-05-20

gRPC 介绍和简单实现

　　gRPC是Google公司基于Protobuf开发的跨语言的开源RPC框架。gRPC基于HTTP/2协议设计，可以基于一个HTTP/2链接提供多个服务，对于移动设备更加友好。本节将讲述gRPC的简单用法。　　最底层为TCP或Unix Socket协议，

极品小肥羊 2020-04-23

gRPC怎样节省您的开发时间

此时，您应该已经听说过" gRPC"。在本文中，我将着重介绍采用gRPC作为微服务之间的通信介质的好处。首先，我将尝试简要介绍一下架构演变的历史。其次，我将重点介绍使用REST和可能出现的问题。最后，我将以我的开发工作流程为例。一切都在

小小书童 2020-04-10

Service Mesh - gRPC 本地联调远程服务

Service Mesh 架构下，服务间调用会通过服务名互相调用，比如在 Kubernetes 、Docker Swarm 集群中，服务 IP 均由集群动态分配，外部网络无法直接访问到集群内部的服务。比较优雅一点的做法是本地启动相应服务，gRPC 调用都连

君小黑 2020-03-04

GRPC协议的相关原理

GRPC的Client与Server，均通过Netty Channel作为数据通信，序列化、反序列化则使用Protobuf，每个请求都将被封装成HTTP2的Stream，在整个生命周期中，客户端Channel应该保持长连接，而不是每次调用重新创建Chann

naiwenw 2020-02-18

使用gRPC-Web从浏览器调用.NET gRPC服务

我很高兴宣布通过.NET对gRPC-Web进行实验性支持。gRPC-Web允许从基于浏览器的应用程序调用gRPC。gRPC-Web通过与HTTP / 1.1和HTTP / 2兼容来解决此问题。gRPC-Web不是一项新技术。有一个稳定的gRPC-Web J

小小书童 2020-02-22

一个新实验：使用gRPC-Web从浏览器调用.NET gRPC服务

今天给大家翻译一篇由ASP.NET首席开发工程师James Newton-King前几天发表的一篇博客，文中带来了一个实验性的产品gRPC-Web。大家可以点击文末的讨论帖进行相关反馈。我会在文章末尾给出原文链接。我很高兴宣布通过.NET对gRPC-Web

dabaoge 2020-02-16

ASP.NET Core 3.0 gRPC拦截器的使用

前面两篇文章给大家介绍了使用gRPC的入门以及双向流的使用，今天介绍的是gRPC中的拦截器。拦截器就像MVC的过滤器或者是ASP.NET Core middleware 一样，具有面向切面的思想，可以在调用服务的时候进行一些统一处理， 很适合在这里处理验证

小小书童 2019-12-23

ASP.NET Core 3.0使用gRPC的具体方法

gRPC ±Googleè·¨èè¨é§èè¨è¨è°¨RPC gRPC·¨¨é°èèéè¨HTTP/2éè¨ Protocol Buffers è。Grpc.AspNetCore ¨¨ASP.NET CoreègRPC° gRPCASP.NET Core

dabaoge 2020-01-25

gRPC - macOS missing ALPN support.

请将 Kestrel 和 gRPC 客户端配置为在不使用 TLS 的情况下使用 HTTP/2。只应在开发过程中执行此操作。如果不使用 TLS，将会在不加密的情况下发送 gRPC 消息。如果没有 TLS，与端点的所有连接默认为 HTTP/1.1，并且 gRP

MAC2007 2020-01-17

window64+mingw+grpc（一）

网上很多都是window+vs+grpc，看了很多帖子都无法通过mingw编译grpc，遇到的问题罗列一下

dabaoge 2020-01-13

在Python中使用gRPC的方法示例【h】

使用Protocol Buffers的跨平台RPC系统。gRPC由两个部分构成，grpcio 和 gRPC 工具， 后者是编译 protocol buffer 以及提供生成代码的插件。使用 gRPC 首先需要做的是设计 protocol buffer。以上

duanlove技术路途 2019-12-28

Go gRPC 调试工具

最近这段时间工作挺忙的，发现已经 3 周没更文了...感谢你们还在，今天给大家分享一款 gRPC 的调试工具。当我们在写 HTTP 接口的时候，使用的是 Postman 进行接口调试，那么在写 gRPC 接口的时候，有没有类似于 Postman 的调试工具

ltstud 2020-01-10
编程语言与高级语言虚拟机杂谈仮

编程语言与高级语言虚拟机杂谈仮

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号