Subscriber.stopAsync() 导致 RejectedExecutionException答案

【问题标题】：Subscriber.stopAsync() results in RejectedExecutionExceptionSubscriber.stopAsync() 导致 RejectedExecutionException
【发布时间】：2017-05-04 15:10:58
【问题描述】：

我的代码基本上遵循官方教程，主要目的是从一个订阅（Constants.UNFINISHEDSUBID）收集所有消息，然后在另一个订阅上重新发布。但目前我面临一个问题，我无法解决。在我的实现中调用subscriber.stopAsync() 会导致以下异常：

Mai 04, 2017 4:59:25 PM com.google.common.util.concurrent.AbstractFuture executeListener
SCHWERWIEGEND: RuntimeException while executing runnable com.google.common.util.concurrent.Futures$6@6e13e898 with executor java.util.concurrent.Executors$DelegatedScheduledExecutorService@2f3c6ac4
java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@60d40af2 rejected from java.util.concurrent.ScheduledThreadPoolExecutor@d55b6e[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 320]
    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 com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:817)
    at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:753)
    at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:613)
    at io.grpc.stub.ClientCalls$GrpcFuture.set(ClientCalls.java:458)
    at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:437)
    at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:428)
    at io.grpc.internal.ClientCallImpl.access$100(ClientCallImpl.java:76)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.close(ClientCallImpl.java:514)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.access$700(ClientCallImpl.java:431)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:546)
    at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:52)
    at io.grpc.internal.SerializingExecutor$TaskRunner.run(SerializingExecutor.java:152)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

我还注意到这种随机的，有时是所有消息，有时只是几条或没有一条被收集。调用subscriber.stopAsync() 方法不正确吗？

我目前的实现：

protected void pullUnfinished() throws Exception {
    List<PubsubMessage> jobsToRepublish = new ArrayList<>();
    SubscriptionName subscription =
            SubscriptionName.create(Constants.PROJECTID, Constants.UNFINISHEDSUBID);

    MessageReceiver receiver = new MessageReceiver() {
        @Override
        public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
            synchronized(jobsToRepublish){
                jobsToRepublish.add(message);
            }
            String unfinishedJob = message.getData().toStringUtf8();
            LOG.info("got message: {}", unfinishedJob);
            consumer.ack();
        }
    };

    Subscriber subscriber = null;
    try {
        ChannelProvider channelProvider = new PlainTextChannelProvider();
        subscriber = Subscriber.defaultBuilder(subscription, receiver)
                               .setChannelProvider(channelProvider)
                               .build();
        subscriber.addListener(new Subscriber.Listener() {
            @Override
            public void failed(Subscriber.State from, Throwable failure) {
                System.err.println(failure);
            }
        }, MoreExecutors.directExecutor());
        subscriber.startAsync().awaitRunning();
        Thread.sleep(60000);
    } finally {
        if (subscriber != null) {
            subscriber.stopAsync(); //Causes the exception
        }
    }
    publishJobs(jobsToRepublish);
}

public class PlainTextChannelProvider implements ChannelProvider {

    @Override
    public boolean shouldAutoClose() {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public boolean needsExecutor() {
        // TODO Auto-generated method stub
        return false;
    }

      @Override
      public ManagedChannel getChannel() throws IOException {
        return NettyChannelBuilder.forAddress("localhost", 8085)
          .negotiationType(NegotiationType.PLAINTEXT)
          .build();
      }

      @Override
      public ManagedChannel getChannel(Executor executor) throws IOException {
        return getChannel();
      } 
}

【问题讨论】：

尝试添加停止信号。请参阅startSync 中的示例代码以了解它是如何实现的。同样如相关SO post 中所述，RejectedExecutionException 是由队列已满导致您无法添加更多线程或 ThreadPool 已关闭。检查你的代码实现。希望这会有所帮助。
已经试过了，对我没有任何改变我仍然得到同样的rejectedExecutionException。我尝试了这些snippets 并仍然收到异常，但似乎我收到了消息，尽管遇到异常让我很恼火，还有其他想法吗？
我遇到了与subscriber.stopAsync() 完全相同的问题；使用最新的库，你发现了吗？

标签： java google-cloud-messaging gcloud google-cloud-pubsub

【解决方案1】：

从 JUnit 测试中运行类似代码时，我遇到了完全相同的问题，并且在一般多线程中发现了这个 related answer，这表明 ThreadPool 已关闭，而侦听器仍在引用它。我也查看了GitHub上Subscriber.java的代码，在javaDoc中找到了startAsync()接收多条消息的例子，建议等待stopAsync()终止。

尝试改变

subscriber.stopAsync();

到

subscriber.stopAsync().awaitTerminated();

为我工作。

【讨论】：