我在kotlin用协同程序制作了一个简单的grpc服务器,用java制作了一个客户端。在客户机中,我启用并配置了重试策略,但它不起作用。我花了很多时间寻找解决方案,相信我的客户机坏了,但问题出在服务器上。我会给你看代码。
这是我的原始文件:
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()
}
}
问题是,怎么了?有人能帮我吗?
1条答案
按热度按时间xzlaal3s1#
该代码是否由grpc生成:
grpc依赖于
StreamObserver
在呼叫后向客户端发送响应responseObserver.onCompleted()
或responseObserver.onError
,请确保您的代码可以正常工作。