mirror of https://github.com/longtai-cn/hippo4j
mingri31164
6 months ago
parent
1c01362410
commit
2efaa3a56d
@ -1,184 +0,0 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.common.executor.support;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
/**
|
||||
* ThreadPoolRebuilder Concurrency Test
|
||||
* Verifies concurrency control and task migration safety during thread pool rebuilding
|
||||
*/
|
||||
public class ThreadPoolRebuilderConcurrencyTest {
|
||||
|
||||
/**
|
||||
* Scenario 1: Concurrent rebuild requests should be serialized
|
||||
* Verification: when multiple threads attempt to rebuild the same thread pool at the same time,
|
||||
* only one succeeds while others are blocked
|
||||
*/
|
||||
@Test
|
||||
public void testConcurrentRebuildSerialize() throws InterruptedException {
|
||||
System.out.println("\n========== Scenario 1: Concurrent rebuild serialization ==========");
|
||||
|
||||
// Note: This test requires the actual ThreadPoolRebuilder class, here it is only a design validation
|
||||
// Actual testing should be executed in the starters/threadpool/server module
|
||||
|
||||
System.out.println("Design validation: ReentrantLock ensures rebuild operations for the same threadPoolId are serialized");
|
||||
System.out.println(" - tryLock() returns immediately to avoid blocking");
|
||||
System.out.println(" - Subsequent requests receive warning logs and return false");
|
||||
}
|
||||
|
||||
/**
|
||||
* Scenario 2: Task migration tracking and error handling
|
||||
* Verification: migration process records success/failure counts,
|
||||
* failures throw exceptions that trigger rollback
|
||||
*/
|
||||
@Test
|
||||
public void testTaskTransferTracking() {
|
||||
System.out.println("\n========== Scenario 2: Task migration tracking ==========");
|
||||
|
||||
ThreadPoolExecutor fromPool = new ThreadPoolExecutor(
|
||||
2, 4, 60, TimeUnit.SECONDS,
|
||||
new LinkedBlockingQueue<>(100));
|
||||
|
||||
ThreadPoolExecutor toPool = new ThreadPoolExecutor(
|
||||
2, 4, 60, TimeUnit.SECONDS,
|
||||
new ArrayBlockingQueue<>(100));
|
||||
|
||||
// Submit 10 tasks to source pool
|
||||
for (int i = 0; i < 10; i++) {
|
||||
fromPool.execute(() -> {
|
||||
try {
|
||||
Thread.sleep(5000); // Simulate long running task
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Verify tasks in queue
|
||||
int queuedTasks = fromPool.getQueue().size();
|
||||
System.out.println("Tasks in source pool queue: " + queuedTasks);
|
||||
Assert.assertTrue("Queue should contain tasks", queuedTasks > 0);
|
||||
|
||||
System.out.println("Design validation: migration process records transferredCount and failedCount");
|
||||
System.out.println(" - Successful transfers logged with count");
|
||||
System.out.println(" - Failures throw exceptions and trigger rollback");
|
||||
|
||||
fromPool.shutdownNow();
|
||||
toPool.shutdownNow();
|
||||
}
|
||||
|
||||
/**
|
||||
* Scenario 3: Rollback old pool state check
|
||||
* Verification: after rollback, old pool termination status is checked
|
||||
* and warnings are logged for unfinished tasks
|
||||
*/
|
||||
@Test
|
||||
public void testRollbackAndOldPoolStatus() {
|
||||
System.out.println("\n========== Scenario 3: Rollback state check ==========");
|
||||
|
||||
System.out.println("Design validation: rollback checks old pool state");
|
||||
System.out.println(" - Call oldExecutor.isTerminated() to verify termination");
|
||||
System.out.println(" - If not terminated, log warnings with activeCount");
|
||||
System.out.println(" - Prevent resource leaks from unfinished tasks");
|
||||
}
|
||||
|
||||
/**
|
||||
* Scenario 4: Atomicity of registry switch
|
||||
* Verification: switchRegistry operation runs under lock protection
|
||||
* to prevent concurrent overwrites
|
||||
*/
|
||||
@Test
|
||||
public void testRegistrySwitchAtomicity() {
|
||||
System.out.println("\n========== Scenario 4: Registry switch atomicity ==========");
|
||||
|
||||
System.out.println("Design validation: registry switch runs under ReentrantLock protection");
|
||||
System.out.println(" - switchRegistry invoked inside doRebuildAndSwitch");
|
||||
System.out.println(" - Entire doRebuildAndSwitch runs under lock");
|
||||
System.out.println(" - Prevent inconsistent registry state due to concurrent rebuilds");
|
||||
}
|
||||
|
||||
/**
|
||||
* Scenario 5: New task submission during migration
|
||||
* Verification: after registry switch, new tasks are routed to the new pool
|
||||
*/
|
||||
@Test
|
||||
public void testNewTasksDuringMigration() throws InterruptedException {
|
||||
System.out.println("\n========== Scenario 5: New task routing during migration ==========");
|
||||
|
||||
ThreadPoolExecutor oldPool = new ThreadPoolExecutor(
|
||||
2, 4, 60, TimeUnit.SECONDS,
|
||||
new LinkedBlockingQueue<>(10));
|
||||
|
||||
ThreadPoolExecutor newPool = new ThreadPoolExecutor(
|
||||
2, 4, 60, TimeUnit.SECONDS,
|
||||
new ArrayBlockingQueue<>(10));
|
||||
|
||||
CountDownLatch latch = new CountDownLatch(5);
|
||||
AtomicInteger oldPoolTasks = new AtomicInteger(0);
|
||||
AtomicInteger newPoolTasks = new AtomicInteger(0);
|
||||
|
||||
// Submit tasks to old pool
|
||||
for (int i = 0; i < 3; i++) {
|
||||
oldPool.execute(() -> {
|
||||
oldPoolTasks.incrementAndGet();
|
||||
latch.countDown();
|
||||
});
|
||||
}
|
||||
|
||||
// Simulate registry switch: new tasks submitted to new pool
|
||||
for (int i = 0; i < 2; i++) {
|
||||
newPool.execute(() -> {
|
||||
newPoolTasks.incrementAndGet();
|
||||
latch.countDown();
|
||||
});
|
||||
}
|
||||
|
||||
latch.await(2, TimeUnit.SECONDS);
|
||||
|
||||
System.out.println("Tasks executed by old pool: " + oldPoolTasks.get());
|
||||
System.out.println("Tasks executed by new pool: " + newPoolTasks.get());
|
||||
Assert.assertEquals("Old pool should execute 3 tasks", 3, oldPoolTasks.get());
|
||||
Assert.assertEquals("New pool should execute 2 tasks", 2, newPoolTasks.get());
|
||||
|
||||
System.out.println("Test passed: after registry switch, new tasks are routed to the new pool");
|
||||
|
||||
oldPool.shutdownNow();
|
||||
newPool.shutdownNow();
|
||||
}
|
||||
|
||||
/**
|
||||
* Scenario 6: Rollback protection on migration failure
|
||||
* Verification: when migration fails, registry is rolled back and new pool shut down
|
||||
*/
|
||||
@Test
|
||||
public void testRollbackOnTransferFailure() {
|
||||
System.out.println("\n========== Scenario 6: Migration failure rollback ==========");
|
||||
|
||||
System.out.println("Design validation: rollback logic on migration failure");
|
||||
System.out.println(" - transferSuccess flag controls rollback");
|
||||
System.out.println(" - On failure: ThreadPoolExecutorRegistry.putHolder(oldHolder)");
|
||||
System.out.println(" - On failure: safeShutdownNow(newExecutor)");
|
||||
System.out.println(" - Return false to notify caller of rebuild failure");
|
||||
System.out.println(" - Old pool continues serving without business interruption");
|
||||
}
|
||||
}
|
||||
@ -1,204 +0,0 @@
|
||||
/*
|
||||
* 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 cn.hippo4j.springboot.starter.core;
|
||||
|
||||
import cn.hippo4j.common.executor.ThreadPoolExecutorHolder;
|
||||
import cn.hippo4j.common.executor.ThreadPoolExecutorRegistry;
|
||||
import cn.hippo4j.common.executor.support.BlockingQueueTypeEnum;
|
||||
import cn.hippo4j.core.executor.DynamicThreadPoolExecutor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
/**
|
||||
* Thread-pool rebuild and switch helper.
|
||||
*/
|
||||
@Slf4j
|
||||
public final class ThreadPoolRebuilder {
|
||||
|
||||
private static final Map<String, ReentrantLock> REBUILD_LOCKS = new ConcurrentHashMap<>();
|
||||
|
||||
private ThreadPoolRebuilder() {
|
||||
}
|
||||
|
||||
public static boolean rebuildAndSwitch(ThreadPoolExecutor oldExecutor,
|
||||
Integer newQueueType,
|
||||
Integer capacity,
|
||||
String threadPoolId) {
|
||||
if (oldExecutor == null || newQueueType == null || threadPoolId == null) {
|
||||
return false;
|
||||
}
|
||||
ReentrantLock lock = REBUILD_LOCKS.computeIfAbsent(threadPoolId, k -> new ReentrantLock());
|
||||
if (!lock.tryLock()) {
|
||||
log.warn("Thread pool [{}] is already being rebuilt, skipping concurrent rebuild request.", threadPoolId);
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
return doRebuildAndSwitch(oldExecutor, newQueueType, capacity, threadPoolId);
|
||||
} finally {
|
||||
try {
|
||||
lock.unlock();
|
||||
} finally {
|
||||
REBUILD_LOCKS.remove(threadPoolId, lock);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean doRebuildAndSwitch(ThreadPoolExecutor oldExecutor,
|
||||
Integer newQueueType,
|
||||
Integer capacity,
|
||||
String threadPoolId) {
|
||||
BlockingQueue<Runnable> newQueue = BlockingQueueTypeEnum.createBlockingQueue(newQueueType, capacity);
|
||||
if (newQueue == null) {
|
||||
log.warn("Rebuild skipped. Unable to create queue by type: {}", newQueueType);
|
||||
return false;
|
||||
}
|
||||
int core = oldExecutor.getCorePoolSize();
|
||||
int max = oldExecutor.getMaximumPoolSize();
|
||||
long keepAlive = oldExecutor.getKeepAliveTime(TimeUnit.SECONDS);
|
||||
ThreadFactory factory = oldExecutor.getThreadFactory();
|
||||
RejectedExecutionHandler rejected = oldExecutor.getRejectedExecutionHandler();
|
||||
boolean allowCoreTimeout = oldExecutor.allowsCoreThreadTimeOut();
|
||||
ThreadPoolExecutor newExecutor;
|
||||
if (oldExecutor instanceof DynamicThreadPoolExecutor) {
|
||||
DynamicThreadPoolExecutor dynOld = (DynamicThreadPoolExecutor) oldExecutor;
|
||||
Long executeTimeout = dynOld.getExecuteTimeOut();
|
||||
newExecutor = new DynamicThreadPoolExecutor(
|
||||
core,
|
||||
max,
|
||||
keepAlive,
|
||||
TimeUnit.SECONDS,
|
||||
executeTimeout == null ? 0L : executeTimeout,
|
||||
false,
|
||||
0L,
|
||||
newQueue,
|
||||
threadPoolId,
|
||||
factory,
|
||||
rejected);
|
||||
} else {
|
||||
newExecutor = new ThreadPoolExecutor(core, max, keepAlive, TimeUnit.SECONDS, newQueue, factory, rejected);
|
||||
}
|
||||
newExecutor.allowCoreThreadTimeOut(allowCoreTimeout);
|
||||
try {
|
||||
newExecutor.prestartAllCoreThreads();
|
||||
} catch (Throwable ignore) {
|
||||
}
|
||||
boolean transferSuccess = false;
|
||||
try {
|
||||
transferQueuedTasks(oldExecutor, newExecutor);
|
||||
transferSuccess = true;
|
||||
} catch (Throwable ex) {
|
||||
log.error("Queue transfer failed for thread pool [{}], keeping old executor unchanged.", threadPoolId, ex);
|
||||
}
|
||||
if (!transferSuccess) {
|
||||
safeShutdownNow(newExecutor);
|
||||
return false;
|
||||
}
|
||||
ThreadPoolExecutorHolder oldHolder = switchRegistry(threadPoolId, newExecutor);
|
||||
oldExecutor.shutdown();
|
||||
awaitQuietly(oldExecutor, 500, TimeUnit.MILLISECONDS);
|
||||
if (!oldExecutor.isTerminated()) {
|
||||
log.warn("Old thread pool [{}] did not terminate within 500ms, {} tasks may still be running.",
|
||||
threadPoolId, oldExecutor.getActiveCount());
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static void transferQueuedTasks(ThreadPoolExecutor from, ThreadPoolExecutor to) {
|
||||
BlockingQueue<Runnable> fromQueue = from.getQueue();
|
||||
int transferredCount = 0;
|
||||
int failedCount = 0;
|
||||
try {
|
||||
List<Runnable> batch = new ArrayList<>();
|
||||
fromQueue.drainTo(batch);
|
||||
for (Runnable r : batch) {
|
||||
try {
|
||||
to.execute(r);
|
||||
transferredCount++;
|
||||
} catch (Throwable ex) {
|
||||
failedCount++;
|
||||
try {
|
||||
from.execute(r);
|
||||
log.warn("Failed to transfer task to new executor, restored to old executor.", ex);
|
||||
} catch (Throwable reEnqueueEx) {
|
||||
if (!fromQueue.offer(r)) {
|
||||
log.error("Failed to restore task to old executor, task may be lost.", reEnqueueEx);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (Throwable ex) {
|
||||
log.error("Failed to drain tasks from old queue.", ex);
|
||||
}
|
||||
Runnable task;
|
||||
while ((task = fromQueue.poll()) != null) {
|
||||
try {
|
||||
to.execute(task);
|
||||
transferredCount++;
|
||||
} catch (Throwable ex) {
|
||||
failedCount++;
|
||||
try {
|
||||
from.execute(task);
|
||||
log.warn("Failed to transfer remaining task to new executor, restored to old executor.", ex);
|
||||
} catch (Throwable reEnqueueEx) {
|
||||
if (!fromQueue.offer(task)) {
|
||||
log.error("Failed to restore remaining task to old executor, task may be lost.", reEnqueueEx);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if (transferredCount > 0 || failedCount > 0) {
|
||||
log.info("Task transfer completed: {} tasks transferred, {} tasks failed (restored to old executor).", transferredCount, failedCount);
|
||||
}
|
||||
if (failedCount > 0) {
|
||||
throw new RuntimeException("Task transfer failed: " + failedCount + " tasks could not be transferred.");
|
||||
}
|
||||
}
|
||||
|
||||
private static void safeShutdownNow(ThreadPoolExecutor executor) {
|
||||
try {
|
||||
List<Runnable> dropped = executor.shutdownNow();
|
||||
if (!dropped.isEmpty()) {
|
||||
log.warn("Dropped {} queued tasks during rollback.", dropped.size());
|
||||
}
|
||||
} catch (Throwable ignore) {
|
||||
}
|
||||
}
|
||||
|
||||
private static void awaitQuietly(ThreadPoolExecutor executor, long time, TimeUnit unit) {
|
||||
try {
|
||||
executor.awaitTermination(time, unit);
|
||||
} catch (InterruptedException ignore) {
|
||||
Thread.currentThread().interrupt();
|
||||
} catch (Throwable ignore) {
|
||||
}
|
||||
}
|
||||
|
||||
private static ThreadPoolExecutorHolder switchRegistry(String threadPoolId, ThreadPoolExecutor newExecutor) {
|
||||
Map<String, ThreadPoolExecutorHolder> map = ThreadPoolExecutorRegistry.getHolderMap();
|
||||
ThreadPoolExecutorHolder holder = map.get(threadPoolId);
|
||||
ThreadPoolExecutorHolder newHolder = new ThreadPoolExecutorHolder(threadPoolId, newExecutor,
|
||||
holder == null ? null : holder.getExecutorProperties());
|
||||
ThreadPoolExecutorRegistry.putHolder(newHolder);
|
||||
return holder;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue