You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by yu...@apache.org on 2017/03/06 12:01:23 UTC
incubator-rocketmq git commit: [ROCKETMQ-119] Add ThreadUtils and
shutdown PullMessageService properly
Repository: incubator-rocketmq
Updated Branches:
refs/heads/develop 53b98d0d8 -> e3f4251c9
[ROCKETMQ-119] Add ThreadUtils and shutdown PullMessageService properly
Project: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/commit/e3f4251c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/tree/e3f4251c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/diff/e3f4251c
Branch: refs/heads/develop
Commit: e3f4251c91a73f4e51732bcb1690554ac5fb3096
Parents: 53b98d0
Author: yukon <yu...@apache.org>
Authored: Mon Mar 6 20:01:10 2017 +0800
Committer: yukon <yu...@apache.org>
Committed: Mon Mar 6 20:01:10 2017 +0800
----------------------------------------------------------------------
.../impl/consumer/PullMessageService.java | 7 +
.../rocketmq/common/utils/ThreadUtils.java | 174 +++++++++++++++++++
2 files changed, 181 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/e3f4251c/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullMessageService.java
----------------------------------------------------------------------
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
index 8ddd483..ed4b837 100644
--- 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
@@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.ServiceThread;
+import org.apache.rocketmq.common.utils.ThreadUtils;
import org.slf4j.Logger;
public class PullMessageService extends ServiceThread {
@@ -98,6 +99,12 @@ public class PullMessageService extends ServiceThread {
}
@Override
+ public void shutdown(boolean interrupt) {
+ super.shutdown(interrupt);
+ ThreadUtils.shutdownGracefully(this.scheduledExecutorService, 1000, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
public String getServiceName() {
return PullMessageService.class.getSimpleName();
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/e3f4251c/common/src/main/java/org/apache/rocketmq/common/utils/ThreadUtils.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/utils/ThreadUtils.java b/common/src/main/java/org/apache/rocketmq/common/utils/ThreadUtils.java
new file mode 100644
index 0000000..8c28d70
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/utils/ThreadUtils.java
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.utils;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class ThreadUtils {
+ private static final Logger log = LoggerFactory.getLogger(LoggerName.TOOLS_LOGGER_NAME);
+
+ public static ExecutorService newThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
+ TimeUnit unit, BlockingQueue<Runnable> workQueue, String processName, boolean isDaemon) {
+ return new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, newThreadFactory(processName, isDaemon));
+ }
+
+ public static ExecutorService newSingleThreadExecutor(String processName, boolean isDaemon) {
+ return Executors.newSingleThreadExecutor(newThreadFactory(processName, isDaemon));
+ }
+
+ public static ScheduledExecutorService newSingleThreadScheduledExecutor(String processName, boolean isDaemon) {
+ return Executors.newSingleThreadScheduledExecutor(newThreadFactory(processName, isDaemon));
+ }
+
+ public static ScheduledExecutorService newFixedThreadScheduledPool(int nThreads, String processName,
+ boolean isDaemon) {
+ return Executors.newScheduledThreadPool(nThreads, newThreadFactory(processName, isDaemon));
+ }
+
+ public static ThreadFactory newThreadFactory(String processName, boolean isDaemon) {
+ return newGenericThreadFactory("Remoting-" + processName, isDaemon);
+ }
+
+ public static ThreadFactory newGenericThreadFactory(String processName) {
+ return newGenericThreadFactory(processName, false);
+ }
+
+ public static ThreadFactory newGenericThreadFactory(String processName, int threads) {
+ return newGenericThreadFactory(processName, threads, false);
+ }
+
+ public static ThreadFactory newGenericThreadFactory(final String processName, final boolean isDaemon) {
+ return new ThreadFactory() {
+ private AtomicInteger threadIndex = new AtomicInteger(0);
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread thread = new Thread(r, String.format("%s_%d", processName, this.threadIndex.incrementAndGet()));
+ thread.setDaemon(isDaemon);
+ return thread;
+ }
+ };
+ }
+
+ public static ThreadFactory newGenericThreadFactory(final String processName, final int threads,
+ final boolean isDaemon) {
+ return new ThreadFactory() {
+ private AtomicInteger threadIndex = new AtomicInteger(0);
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread thread = new Thread(r, String.format("%s_%d_%d", processName, threads, this.threadIndex.incrementAndGet()));
+ thread.setDaemon(isDaemon);
+ return thread;
+ }
+ };
+ }
+
+ /**
+ * Create a new thread
+ *
+ * @param name The name of the thread
+ * @param runnable The work for the thread to do
+ * @param daemon Should the thread block JVM stop?
+ * @return The unstarted thread
+ */
+ public static Thread newThread(String name, Runnable runnable, boolean daemon) {
+ Thread thread = new Thread(runnable, name);
+ thread.setDaemon(daemon);
+ thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ public void uncaughtException(Thread t, Throwable e) {
+ log.error("Uncaught exception in thread '" + t.getName() + "':", e);
+ }
+ });
+ return thread;
+ }
+
+ /**
+ * Shutdown passed thread using isAlive and join.
+ *
+ * @param t Thread to stop
+ */
+ public static void shutdownGracefully(final Thread t) {
+ shutdownGracefully(t, 0);
+ }
+
+ /**
+ * Shutdown passed thread using isAlive and join.
+ *
+ * @param millis Pass 0 if we're to wait forever.
+ * @param t Thread to stop
+ */
+ public static void shutdownGracefully(final Thread t, final long millis) {
+ if (t == null)
+ return;
+ while (t.isAlive()) {
+ try {
+ t.interrupt();
+ t.join(millis);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ /**
+ * An implementation of the graceful stop sequence recommended by
+ * {@link ExecutorService}.
+ *
+ * @param executor executor
+ * @param timeout timeout
+ * @param timeUnit timeUnit
+ */
+ public static void shutdownGracefully(ExecutorService executor, long timeout, TimeUnit timeUnit) {
+ // Disable new tasks from being submitted.
+ executor.shutdown();
+ try {
+ // Wait a while for existing tasks to terminate.
+ if (!executor.awaitTermination(timeout, timeUnit)) {
+ executor.shutdownNow();
+ // Wait a while for tasks to respond to being cancelled.
+ if (!executor.awaitTermination(timeout, timeUnit)) {
+ log.warn(String.format("%s didn't terminate!", executor));
+ }
+ }
+ } catch (InterruptedException ie) {
+ // (Re-)Cancel if current thread also interrupted.
+ executor.shutdownNow();
+ // Preserve interrupt status.
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * A constructor to stop this class being constructed.
+ */
+ private ThreadUtils() {
+ // Unused
+
+ }
+}