Check if pull message service has shutdown before scheduling pull requests (apache#277)

lizhanhui · web-flow · commit bf848c14e1a9 · 2018-04-08T16:32:31.000+08:00
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullMessageService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullMessageService.java
@@ -44,13 +44,16 @@ public PullMessageService(MQClientInstance mQClientFactory) {
     }
 
     public void executePullRequestLater(final PullRequest pullRequest, final long timeDelay) {
-        this.scheduledExecutorService.schedule(new Runnable() {
-
-            @Override
-            public void run() {
-                PullMessageService.this.executePullRequestImmediately(pullRequest);
-            }
-        }, timeDelay, TimeUnit.MILLISECONDS);
+        if (!isStopped()) {
+            this.scheduledExecutorService.schedule(new Runnable() {
+                @Override
+                public void run() {
+                    PullMessageService.this.executePullRequestImmediately(pullRequest);
+                }
+            }, timeDelay, TimeUnit.MILLISECONDS);
+        } else {
+            log.warn("PullMessageServiceScheduledThread has shutdown");
+        }
     }
 
     public void executePullRequestImmediately(final PullRequest pullRequest) {
@@ -62,7 +65,11 @@ public void executePullRequestImmediately(final PullRequest pullRequest) {
     }
 
     public void executeTaskLater(final Runnable r, final long timeDelay) {
-        this.scheduledExecutorService.schedule(r, timeDelay, TimeUnit.MILLISECONDS);
+        if (!isStopped()) {
+            this.scheduledExecutorService.schedule(r, timeDelay, TimeUnit.MILLISECONDS);
+        } else {
+            log.warn("PullMessageServiceScheduledThread has shutdown");
+        }
     }
 
     public ScheduledExecutorService getScheduledExecutorService() {
@@ -86,10 +93,8 @@ public void run() {
         while (!this.isStopped()) {
             try {
                 PullRequest pullRequest = this.pullRequestQueue.take();
-                if (pullRequest != null) {
-                    this.pullMessage(pullRequest);
-                }
-            } catch (InterruptedException e) {
+                this.pullMessage(pullRequest);
+            } catch (InterruptedException ignored) {
             } catch (Exception e) {
                 log.error("Pull Message Service Run Method exception", e);
             }
