Scala Grpc 失败，错误 io.grpc.StatusRuntimeException: 已取消: 无法读取消息

Question

我正在尝试使用 GRPC 在 Scala 中编写流服务。为此，我写了这个原型文件

syntax = "proto3";
package com.abhi.grpc;
message TimeRequest{}
message TimeResponse {
int64 currentTime = 1;
}
service Clock {
rpc StreamTime(TimeRequest) returns (stream TimeResponse);
}

这是我的服务器端代码

import com.abhi.grpc.clock.{ClockGrpc, TimeRequest, TimeResponse}
import io.grpc.stub.StreamObserver
import monix.execution.Scheduler
import monix.execution.Scheduler.{global => scheduler}
import scala.concurrent.duration._
object ClockGrpcServer extends GrpcServer with App {
val ssd = ClockGrpc.bindService(new ClockGRPC(), Scheduler.global)
runServer(ssd, "Clock")
}
class ClockGRPC extends ClockGrpc.Clock {
override def streamTime(request: TimeRequest, responseObserver: StreamObserver[TimeResponse]): Unit = {
scheduler.scheduleWithFixedDelay(0.seconds, 3.seconds) {
responseObserver.onNext(TimeResponse(System.currentTimeMillis))
}
}
}

这是我的客户

object ClockGrpcClient extends App {
val channel = ManagedChannelBuilder.forAddress("localhost", 50051).usePlaintext(true).build()
val stub = ClockGrpc.stub(channel)
val observer = new StreamObserver[TimeResponse] {
override def onError(t: Throwable): Unit = println(s"failed with error ${t}")
override def onCompleted(): Unit = println("closing observer")
override def onNext(value: TimeResponse): Unit = println(s"received time ${new DateTime(value)}")
}
stub.streamTime(TimeRequest(), observer)
StdIn.readLine()
}

当我运行服务器和客户端时。服务器在收到来自客户端的任何消息后立即引发以下错误

io.grpc.StatusRuntimeException: CANCELLED
at io.grpc.Status.asRuntimeException(Status.java:534)
at io.grpc.stub.ServerCalls$ServerCallStreamObserverImpl.onNext(ServerCalls.java:279)
at com.abhi.ClockGRPC.$anonfun$streamTime$1(ClockGRPC.scala:22)
at monix.execution.internal.RunnableAction.run(RunnableAction.scala:25)
at monix.execution.schedulers.ReferenceScheduler$$anon$1.run(ReferenceScheduler.scala:45)
at scala.concurrent.impl.ExecutionContextImpl$AdaptedForkJoinTask.exec(ExecutionContextImpl.scala:140)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

我用谷歌搜索了一下，发现了这篇文章

https://blog.codecentric.de/en/2017/01/hello-grpc-scalapb/

基于此，我将服务器更改为使用 java.util 调度程序

class ClockGRPC extends ClockGrpc.Clock {
val scheduler = Executors.newSingleThreadScheduledExecutor()
override def streamTime(request: TimeRequest, responseObserver: StreamObserver[TimeResponse]): Unit = {
val tick = new Runnable {
val counter = new AtomicInteger(10)
def run() =
if (counter.getAndDecrement() >= 0) {
val currentTime = System.currentTimeMillis()
responseObserver.onNext(TimeResponse(currentTime))
} else {
scheduler.shutdown()
responseObserver.onCompleted()
}
}
scheduler.scheduleAtFixedRate(tick, 0l, 3000l, TimeUnit.SECONDS)
}
}

但我仍然收到取消错误。所以我无法让流示例工作。

Answer 1

我几乎放弃了这个问题。但是今天回来解决了它。

问题出在生产线上

override def onNext(value: TimeResponse): Unit = println(s"received time ${new DateTime(value)}")

值不能传递给new DateTime

让事情雪上加霜。 如果回调方法中发生异常。Grpc 吞下它并用一般错误消息替换它

info] Running com.abhi.ClockGrpcClient failed with error io.grpc.StatusRuntimeException: CANCELLED: Failed to read message.

我的运气不好，他 DateTime 使用对象作为参数，因此编译成功，但调用在运行时失败，异常被 Grpc 吞噬。

我把它留在这里，以便它帮助其他人。

[info] Running com.abhi.ClockGrpcClient failed with error io.grpc.StatusRuntimeException: CANCELLED: Failed to read message

表示某人在回调函数中出错。