Google /sub com.google.cloud.pubsub.v1.StreamingSubscriberConnection

Question

我有一个雪犁丰富了她在GKE中托管的应用程序，它消耗了来自google发布/订阅的消息，而充实她的应用程序正在抛出下面的错误。

我可以在发布/订阅订阅中看到num_undelivered_messages计数峰值(超过50000)，每天3-4次，我认为这些错误消息的出现是因为充实她的应用程序无法从上述订阅中获取消息。

为什么应用程序有时无法连接到发布/订阅？

任何帮助都是非常感谢的。

Apr 12, 2022 12:30:32 PM com.google.cloud.pubsub.v1.StreamingSubscriberConnection$2 onFailure
WARNING: failed to send operations
com.google.api.gax.rpc.UnavailableException: io.grpc.StatusRuntimeException: UNAVAILABLE: 502:Bad Gateway
at com.google.api.gax.rpc.ApiExceptionFactory.createException(ApiExceptionFactory.java:69)
at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:72)
at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:60)
at com.google.api.gax.grpc.GrpcExceptionCallable$ExceptionTransformingFuture.onFailure(GrpcExceptionCallable.java:97)
at com.google.api.core.ApiFutures$1.onFailure(ApiFutures.java:68)
at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1050)
at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1176)
at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:969)
at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:760)
at io.grpc.stub.ClientCalls$GrpcFuture.setException(ClientCalls.java:545)
at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:515)
at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:426)
at io.grpc.internal.ClientCallImpl.access$500(ClientCallImpl.java:66)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.close(ClientCallImpl.java:689)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.access$900(ClientCallImpl.java:577)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:751)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:740)
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: io.grpc.StatusRuntimeException: UNAVAILABLE: 502:Bad Gateway
at io.grpc.Status.asRuntimeException(Status.java:533)
... 15 more

Answer 1

订阅中的邮件积累表明您的订阅服务器没有跟上消息流。

要监视订阅者，您可以创建一个仪表板，其中包含待办事项度量：num_undelivered_messages和oldest_unacked_message_age (订阅的待办事项中最古老的未加加消息的年龄)，它们按资源聚合，用于所有订阅。

• --如果oldest_unacked_message和num_undelivered_messages都在增长，那是因为订阅者跟不上消息量。

解决方案：添加更多订阅者线程/机器，并在代码中查找可能阻止确认消息的bug。

• 如果有一个稳定的、小的积压大小和一个稳定增长的oldest_unacked_message_age，那么可能会有少量无法处理的消息。这可能是因为信息被卡住了。

解决方案：检查应用程序日志以了解某些消息是否导致代码崩溃。这是不太可能的--但可能是-that--冒犯的消息被卡在Pub/Sub上，而不是在您的客户端。

如果

• 超过订阅的消息保留时间，则数据丢失的可能性很高；在这种情况下，最好的选择是在订阅的消息保留时间过期之前设置触发警报。