Skip to content

Commit

Permalink
Fix bug in SofaThreadPoolTaskExecutor threadPoolName setter (#70)
Browse files Browse the repository at this point in the history
  • Loading branch information
@alaneuler
alaneuler authored Apr 28, 2020
1 parent 5abb1cf commit 64dda70
Show file tree
Hide file tree
Showing 9 changed files with 65 additions and 49 deletions.
2 changes: 1 addition & 1 deletion core/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
<parent>
<artifactId>sofa-common-tools-parent</artifactId>
<groupId>com.alipay.sofa.common</groupId>
<version>1.1.1</version>
<version>1.1.2</version>
</parent>
<modelVersion>4.0.0</modelVersion>

Expand Down
73 changes: 39 additions & 34 deletions core/src/main/java/com/alipay/sofa/common/thread/SofaThreadPoolExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ public class SofaThreadPoolExecutor extends ThreadPoolExecutor implements Runnab
SIMPLE_CLASS_NAME
+ "_SCHEDULER"));

private String name;
private String threadPoolName;

private long taskTimeout = DEFAULT_TASK_TIMEOUT;
private long period = DEFAULT_PERIOD;
Expand All @@ -65,17 +65,18 @@ public class SofaThreadPoolExecutor extends ThreadPoolExecutor implements Runnab
* @param workQueue same as in {@link ThreadPoolExecutor}
* @param threadFactory same as in {@link ThreadPoolExecutor}
* @param handler same as in {@link ThreadPoolExecutor}
* @param name name of this thread pool
* @param threadPoolName name of this thread pool
* @param taskTimeout task execution timeout
* @param period task checking and logging period
* @param timeUnit unit of taskTimeout and period
*/
public SofaThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory, RejectedExecutionHandler handler,
String name, long taskTimeout, long period, TimeUnit timeUnit) {
String threadPoolName, long taskTimeout, long period,
TimeUnit timeUnit) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
this.name = name;
this.threadPoolName = threadPoolName;
this.taskTimeout = taskTimeout;
this.period = period;
this.timeUnit = timeUnit;
Expand All @@ -86,53 +87,54 @@ public SofaThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAl
public SofaThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory, RejectedExecutionHandler handler,
String name) {
String threadPoolName) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler,
name, DEFAULT_TASK_TIMEOUT, DEFAULT_PERIOD, DEFAULT_TIME_UNIT);
threadPoolName, DEFAULT_TASK_TIMEOUT, DEFAULT_PERIOD, DEFAULT_TIME_UNIT);
}

public SofaThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue, String name) {
TimeUnit unit, BlockingQueue<Runnable> workQueue,
String threadPoolName) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
this.name = name;
this.threadPoolName = threadPoolName;
scheduleAndRegister(period, timeUnit);
}

public SofaThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
name = createName();
threadPoolName = createName();
scheduleAndRegister(period, timeUnit);
}

public SofaThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
name = createName();
threadPoolName = createName();
scheduleAndRegister(period, timeUnit);
}

public SofaThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
name = createName();
threadPoolName = createName();
scheduleAndRegister(period, timeUnit);
}

public SofaThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
name = createName();
threadPoolName = createName();
scheduleAndRegister(period, timeUnit);
}

@Override
protected void terminated() {
super.terminated();
ThreadPoolGovernor.unregisterThreadPoolExecutor(name);
ThreadPoolGovernor.unregisterThreadPoolExecutor(threadPoolName);
synchronized (monitor) {
if (scheduledFuture != null) {
scheduledFuture.cancel(true);
Expand All @@ -151,7 +153,8 @@ private void scheduleAndRegister(long period, TimeUnit unit) {

synchronized (monitor) {
scheduledFuture = scheduler.scheduleAtFixedRate(this, period, period, unit);
ThreadLogger.info("Thread pool '{}' started with period: {} {}", name, period, unit);
ThreadLogger.info("Thread pool '{}' started with period: {} {}", threadPoolName,
period, unit);
}
}

Expand All @@ -163,11 +166,11 @@ public synchronized void startSchedule() {
synchronized (monitor) {
if (scheduledFuture == null) {
scheduledFuture = scheduler.scheduleAtFixedRate(this, period, period, timeUnit);
ThreadLogger.info("Thread pool '{}' started with period: {} {}", name, period,
timeUnit);
} else {
ThreadLogger.warn("Thread pool '{}' is already started with period: {} {}", name,
ThreadLogger.info("Thread pool '{}' started with period: {} {}", threadPoolName,
period, timeUnit);
} else {
ThreadLogger.warn("Thread pool '{}' is already started with period: {} {}",
threadPoolName, period, timeUnit);
}
}
}
Expand All @@ -177,9 +180,9 @@ public void stopSchedule() {
if (scheduledFuture != null) {
scheduledFuture.cancel(true);
scheduledFuture = null;
ThreadLogger.info("Thread pool '{}' stopped.", name);
ThreadLogger.info("Thread pool '{}' stopped.", threadPoolName);
} else {
ThreadLogger.warn("Thread pool '{}' is not scheduling!", name);
ThreadLogger.warn("Thread pool '{}' is not scheduling!", threadPoolName);
}
}
}
Expand All @@ -189,8 +192,8 @@ private void reschedule(long period, TimeUnit unit) {
if (scheduledFuture != null) {
scheduledFuture.cancel(true);
scheduledFuture = scheduler.scheduleAtFixedRate(this, period, period, unit);
ThreadLogger.info("Reschedule thread pool '{}' with period: {} {}", name, period,
unit);
ThreadLogger.info("Reschedule thread pool '{}' with period: {} {}", threadPoolName,
period, unit);
}
}
}
Expand Down Expand Up @@ -230,7 +233,7 @@ public void run() {
ThreadLogger
.warn(
"Task {} in thread pool {} started on {}{} exceeds the limit of {} execution time with stack trace:\n {}",
task, getName(),
task, getThreadPoolName(),
DATE_FORMAT.format(executionInfo.getTaskKickOffTime()),
traceId == null ? "" : " with traceId " + traceId, getTaskTimeout()
+ getTimeUnit()
Expand All @@ -241,11 +244,12 @@ task, getName(),
}

// threadPoolName, #queue, #executing, #idle, #pool, #decayed
ThreadLogger.info("Thread pool '{}' info: [{},{},{},{},{}]", getName(), this.getQueue()
.size(), executingTasks.size(), this.getPoolSize() - executingTasks.size(), this
.getPoolSize(), decayedTaskCount);
ThreadLogger.info("Thread pool '{}' info: [{},{},{},{},{}]", getThreadPoolName(), this
.getQueue().size(), executingTasks.size(),
this.getPoolSize() - executingTasks.size(), this.getPoolSize(), decayedTaskCount);
} catch (Throwable e) {
ThreadLogger.warn("ThreadPool '{}' is interrupted when running: {}", this.name, e);
ThreadLogger.warn("ThreadPool '{}' is interrupted when running: {}",
this.threadPoolName, e);
}
}

Expand Down Expand Up @@ -279,14 +283,14 @@ protected String traceIdSafari(Thread t) {
return null;
}

public String getName() {
return name;
public String getThreadPoolName() {
return threadPoolName;
}

public void setName(String name) {
ThreadPoolGovernor.unregisterThreadPoolExecutor(this.name);
this.name = name;
ThreadPoolGovernor.registerThreadPoolExecutor(name, this);
public void setThreadPoolName(String threadPoolName) {
ThreadPoolGovernor.unregisterThreadPoolExecutor(this.threadPoolName);
this.threadPoolName = threadPoolName;
ThreadPoolGovernor.registerThreadPoolExecutor(threadPoolName, this);
}

public void setPeriod(long period) {
Expand All @@ -301,7 +305,8 @@ public long getTaskTimeout() {
public void setTaskTimeout(long taskTimeout) {
this.taskTimeout = taskTimeout;
this.taskTimeoutMilli = timeUnit.toMillis(taskTimeout);
ThreadLogger.info("Updated '{}' taskTimeout to {} {}", name, taskTimeout, timeUnit);
ThreadLogger.info("Updated '{}' taskTimeout to {} {}", threadPoolName, taskTimeout,
timeUnit);
}

public TimeUnit getTimeUnit() {
Expand Down
17 changes: 13 additions & 4 deletions core/src/main/java/com/alipay/sofa/common/thread/SofaThreadPoolTaskExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ public class SofaThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
protected static TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;

protected SofaThreadPoolExecutor sofaThreadPoolExecutor;
protected String threadPoolName;

@Override
protected ExecutorService initializeExecutor(ThreadFactory threadFactory,
Expand All @@ -49,10 +50,15 @@ protected ExecutorService initializeExecutor(ThreadFactory threadFactory,

SofaThreadPoolExecutor executor;

// When used as Spring bean, setter method is called before init method
if (threadPoolName == null) {
threadPoolName = createName();
}

if (taskDecorator != null) {
executor = new SofaThreadPoolExecutor(getCorePoolSize(), getMaxPoolSize(),
getKeepAliveSeconds(), TimeUnit.SECONDS, queue, threadFactory,
rejectedExecutionHandler, createName(), DEFAULT_TASK_TIMEOUT, DEFAULT_PERIOD,
rejectedExecutionHandler, threadPoolName, DEFAULT_TASK_TIMEOUT, DEFAULT_PERIOD,
DEFAULT_TIME_UNIT) {
@Override
public void execute(Runnable command) {
Expand All @@ -62,7 +68,7 @@ public void execute(Runnable command) {
} else {
executor = new SofaThreadPoolExecutor(getCorePoolSize(), getMaxPoolSize(),
getKeepAliveSeconds(), TimeUnit.SECONDS, queue, threadFactory,
rejectedExecutionHandler, createName(), DEFAULT_TASK_TIMEOUT, DEFAULT_PERIOD,
rejectedExecutionHandler, threadPoolName, DEFAULT_TASK_TIMEOUT, DEFAULT_PERIOD,
DEFAULT_TIME_UNIT);
}

Expand All @@ -81,11 +87,14 @@ protected String createName() {
}

public String getThreadPoolName() {
return sofaThreadPoolExecutor.getName();
return threadPoolName;
}

public void setThreadPoolName(String threadPoolName) {
sofaThreadPoolExecutor.setName(threadPoolName);
this.threadPoolName = threadPoolName;
if (sofaThreadPoolExecutor != null) {
sofaThreadPoolExecutor.setThreadPoolName(threadPoolName);
}
}

public void setTaskTimeout(long taskTimeout) {
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/java/com/alipay/sofa/common/thread/ThreadPoolGovernor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ public static void registerThreadPoolExecutor(String name, ThreadPoolExecutor th
}

public static void registerThreadPoolExecutor(SofaThreadPoolExecutor threadPoolExecutor) {
registerThreadPoolExecutor(threadPoolExecutor.getName(), threadPoolExecutor);
registerThreadPoolExecutor(threadPoolExecutor.getThreadPoolName(), threadPoolExecutor);
}

public static void unregisterThreadPoolExecutor(String name) {
Expand Down
2 changes: 1 addition & 1 deletion log-sofa-boot-starter/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
<parent>
<artifactId>sofa-common-tools-parent</artifactId>
<groupId>com.alipay.sofa.common</groupId>
<version>1.1.1</version>
<version>1.1.2</version>
</parent>
<modelVersion>4.0.0</modelVersion>

Expand Down
4 changes: 2 additions & 2 deletions pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

<groupId>com.alipay.sofa.common</groupId>
<artifactId>sofa-common-tools-parent</artifactId>
<version>1.1.1</version>
<version>1.1.2</version>
<modules>
<module>core</module>
<module>log-sofa-boot-starter</module>
Expand All @@ -27,7 +27,7 @@
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<logback.version>1.1.7</logback.version>
<slf4j.version>1.7.21</slf4j.version>
<common.tools.version>1.1.1</common.tools.version>
<common.tools.version>1.1.2</common.tools.version>
<log4j2.version>2.3</log4j2.version>
<java.version>1.6</java.version>
</properties>
Expand Down
2 changes: 1 addition & 1 deletion test/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
<parent>
<artifactId>sofa-common-tools-parent</artifactId>
<groupId>com.alipay.sofa.common</groupId>
<version>1.1.1</version>
<version>1.1.2</version>
</parent>
<modelVersion>4.0.0</modelVersion>

Expand Down
2 changes: 1 addition & 1 deletion test/src/test/java/com/alipay/sofa/common/thread/SofaThreadPoolExecutorTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ public void testDecayedTask() throws Exception {

@Test
public void testRename() {
threadPool.setName("sofaThreadPoolName");
threadPool.setThreadPoolName("sofaThreadPoolName");
Assert.assertEquals(0, aberrantListAppender.list.size());
Assert.assertEquals(4, infoListAppender.list.size());
Assert.assertTrue(isMatch(getInfoViaIndex(2), INFO,
Expand Down
10 changes: 6 additions & 4 deletions test/src/test/java/com/alipay/sofa/common/thread/ThreadPoolGovernorTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -89,12 +89,14 @@ public void testSofaThreadPoolExecutor() throws Exception {
Assert.assertEquals(9, infoListAppender.list.size());
Assert.assertEquals(0, aberrantListAppender.list.size());
Assert.assertEquals(sofaThreadPoolExecutor1,
ThreadPoolGovernor.getThreadPoolExecutor(sofaThreadPoolExecutor1.getName()));
ThreadPoolGovernor.getThreadPoolExecutor(sofaThreadPoolExecutor1.getThreadPoolName()));
Assert.assertEquals(sofaThreadPoolExecutor2,
ThreadPoolGovernor.getThreadPoolExecutor(sofaThreadPoolExecutor2.getName()));
ThreadPoolGovernor.getThreadPoolExecutor(sofaThreadPoolExecutor2.getThreadPoolName()));

ThreadPoolGovernor.unregisterThreadPoolExecutor(sofaThreadPoolExecutor1.getName());
ThreadPoolGovernor.unregisterThreadPoolExecutor(sofaThreadPoolExecutor2.getName());
ThreadPoolGovernor
.unregisterThreadPoolExecutor(sofaThreadPoolExecutor1.getThreadPoolName());
ThreadPoolGovernor
.unregisterThreadPoolExecutor(sofaThreadPoolExecutor2.getThreadPoolName());
Assert.assertEquals(11, infoListAppender.list.size());
Assert.assertEquals(0, aberrantListAppender.list.size());
sofaThreadPoolExecutor1.shutdownNow();
Expand Down

0 comments on commit 64dda70

Please sign in to comment.