RetryPolicy 不适用于协程

如何解决RetryPolicy 不适用于协程

我在 Kotlin 中使用协程和 Java 制作了一个简单的 gRPC 服务器。在客户端中，我启用并配置了重试策略，但它不起作用。我花了很多时间寻找解决方案，认为我的客户端坏了，但问题出在服务器上。我会告诉你代码。

这是我的原型文件：

Syntax = "proto3";
option java_multiple_files = true;
option java_package = "br.com.will.protoclasses";
option java_outer_classname = "NotificationProto";

package notification;

service Notification {
  rpc SendPush (SendPushNotificationRequest) returns (SendPushNotificationResponse);
}

message SendPushNotificationRequest {
  string title = 1;
  string message = 2;
  string customer_id = 3;
}

message SendPushNotificationResponse {
  string message = 1;
}

这是客户：

open class NotificationClient(private val channel: ManagedChannel) {
    private val stub: NotificationGrpcKt.NotificationCoroutinestub =
        NotificationGrpcKt.NotificationCoroutinestub(channel)

    suspend fun send() {
        val request =
            SendPushNotificationRequest.newBuilder().setCustomerId(UUID.randomUUID().toString()).setMessage("test")
                .setTitle("test").build()
        val response =  stub.sendPush(request)
        println("Received: ${response.message}")
    }

}

suspend fun main(args: Array<String>) {
    val port = System.getenv("PORT")?.toInt() ?: 50051

    val retryPolicy: MutableMap<String,Any> = HashMap()
    retryPolicy["maxAttempts"] = 5.0
    retryPolicy["initialBackoff"] = "10s"
    retryPolicy["maxBackoff"] = "30s"
    retryPolicy["backoffMultiplier"] = 2.0
    retryPolicy["retryableStatusCodes"] = listof<Any>("INTERNAL")

    val methodConfig: MutableMap<String,Any> = HashMap()

    val name: MutableMap<String,Any> = HashMap()
    name["service"] = "notification.Notification"
    name["method"] = "SendPush"
    methodConfig["name"] = listof<Any>(name)
    methodConfig["retryPolicy"] = retryPolicy

    val serviceConfig: MutableMap<String,Any> = HashMap()
    serviceConfig["methodConfig"] = listof<Any>(methodConfig)

    print(serviceConfig)

    val channel = ManagedChannelBuilder.forAddress("localhost",port)
        .usePlaintext()
        .defaultServiceConfig(serviceConfig)
        .enableRetry()
        .build()

    val client = NotificationClient(channel)

    client.send()
}

这是我的 gRPC 服务的一部分，我在其中测试了重试策略（客户端上的重试策略不适用于此实现）：

override suspend fun sendPush(request: SendPushNotificationRequest): SendPushNotificationResponse {
    val count: Int = retryCounter.incrementAndGet()
    log.info("Received a call on method sendPushNotification with payload -> $request")

    if (random.nextFloat() < UNAVAILABLE_PERCENTAGE) {
        log.info("Returning stubbed INTERNAL error. count: $count")
        throw Status.INTERNAL.withDescription("error").asRuntimeException()
    }

    log.info("Returning successful Hello response,count: $count")
    return SendPushNotificationResponse.newBuilder().setMessage("success").build()

}

另一个实现，但现在使用 StreamObserver（这个实现工作正常）：

override fun sendPush(
        request: SendPushNotificationRequest?,responSEObserver: StreamObserver<SendPushNotificationResponse>?
    ) {
        log.info("Received a call on method sendPushNotification with payload -> $request")

        val count: Int = retryCounter.incrementAndGet()
        if (random.nextFloat() < UNAVAILABLE_PERCENTAGE) {
            log.info("Returning stubbed UNAVAILABLE error. count: $count")
            responSEObserver!!.onError(
                Status.UNAVAILABLE.withDescription("error").asRuntimeException()
            )
        } else {
            log.info("Returning successful Hello response,count: $count")

            responSEObserver!!.onNext(SendPushNotificationResponse.newBuilder().setMessage("success").build())
            return responSEObserver.onCompleted()
        }
    }

问题是，怎么了？有人可以帮我吗？

解决方法

这段代码是不是gRPC生成的：

sendPush(request: SendPushNotificationRequest): SendPushNotificationResponse

gRPC 依赖于 StreamObserver 在调用 responseObserver.onCompleted() 或 responseObserver.onError 后向客户端发送响应，请确保您的代码可以正常工作。