2

我使用 GCloud 文档中的示例片段作为订阅者接收 msg。我的 pubsub gcloud jar 版本是 0.19.0-alpha

问题是我可以接收带有属性映射的味精，但我一直有这个异常：

2017-07-12 16:52:25,219 [grpc-default-worker-ELG-1-16] WARN  io.netty.util.concurrent.DefaultPromise - An exception was thrown by io.grpc.netty.NettyClientHandler$3.operationComplete()
java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@fbf4a6d rejected from java.util.concurrent.ScheduledThreadPoolExecutor@25cbe860[Terminated, pool size = 35, active threads = 0, queued tasks = 0, completed tasks = 2403]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:326)
    at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:533)
    at java.util.concurrent.ScheduledThreadPoolExecutor.execute(ScheduledThreadPoolExecutor.java:622)
    at java.util.concurrent.Executors$DelegatedExecutorService.execute(Executors.java:668)
    at io.grpc.internal.SerializingExecutor.execute(SerializingExecutor.java:110)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.onReady(ClientCallImpl.java:573)
    at io.grpc.internal.DelayedStream$DelayedStreamListener.onReady(DelayedStream.java:398)
    at io.grpc.internal.AbstractStream2$TransportState.notifyIfReady(AbstractStream2.java:305)
    at io.grpc.internal.AbstractStream2$TransportState.onStreamAllocated(AbstractStream2.java:248)
    at io.grpc.netty.NettyClientStream$TransportState.setHttp2Stream(NettyClientStream.java:227)
    at io.grpc.netty.NettyClientHandler$3.operationComplete(NettyClientHandler.java:429)
    at io.grpc.netty.NettyClientHandler$3.operationComplete(NettyClientHandler.java:417)
    at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:507)
    at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:481)
    at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:420)
    at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:104)

之后，程序关闭并停止收听和获取 msg。如何解决这种中断，我什至摆脱了具有subscriber.stopAsync() 的finally 子句。

4

1 回答 1

2

他们提供的代码段中有一个错误。您需要在 messaegeIdFuture 上调用 get()。下面的代码解决了这个问题：

Publisher publisher = null;
String projectId = ServiceOptions.getDefaultProjectId();
ProjectTopicName topic = ProjectTopicName.of(projectId, "test");
ApiFuture<String> messageIdFuture = null;
try {
       publisher = Publisher.newBuilder(topic).build();
       ByteString data = ByteString.copyFromUtf8("my-message");
       PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();
       messageIdFuture = publisher.publish(pubsubMessage);
    } catch (IOException e) {
       e.printStackTrace();
    } finally {
        messageIdFuture.get();    //This resolves this issue.
        // Wait on any pending requests
        if (publisher != null) {
            publisher.shutdown();
            //publisher.awaitTermination(1, TimeUnit.SECONDS);
        }
    }
于 2019-11-14T04:23:22.880 回答