java - Subscriber.stopAsync() 导致 RejectedExecutionException

Question

我的代码基本上遵循官方教程，主要目的是收集一个订阅(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();
      } 
}

Answer 1

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

尝试改变

subscriber.stopAsync();

至

subscriber.stopAsync().awaitTerminated();

为我工作。