optimize ThreadPoolRebuilder

pull/1612/head
mingri31164 11 months ago
parent 047f96272b
commit 5e97e3a8e5

@ -55,7 +55,11 @@ public final class ThreadPoolRebuilder {
try { try {
return doRebuildAndSwitch(oldExecutor, newQueueType, capacity, threadPoolId); return doRebuildAndSwitch(oldExecutor, newQueueType, capacity, threadPoolId);
} finally { } finally {
lock.unlock(); try {
lock.unlock();
} finally {
REBUILD_LOCKS.remove(threadPoolId, lock);
}
} }
} }
@ -98,22 +102,18 @@ public final class ThreadPoolRebuilder {
newExecutor.prestartAllCoreThreads(); newExecutor.prestartAllCoreThreads();
} catch (Throwable ignore) { } catch (Throwable ignore) {
} }
ThreadPoolExecutorHolder oldHolder = switchRegistry(threadPoolId, newExecutor);
boolean transferSuccess = false; boolean transferSuccess = false;
try { try {
transferQueuedTasks(oldExecutor, newExecutor); transferQueuedTasks(oldExecutor, newExecutor);
transferSuccess = true; transferSuccess = true;
} catch (Throwable ex) { } catch (Throwable ex) {
log.error("Queue transfer failed for thread pool [{}], attempting to rollback.", threadPoolId, ex); log.error("Queue transfer failed for thread pool [{}], keeping old executor unchanged.", threadPoolId, ex);
} }
if (!transferSuccess) { if (!transferSuccess) {
if (oldHolder != null) {
ThreadPoolExecutorRegistry.putHolder(oldHolder);
log.info("Rolled back to old thread pool [{}]", threadPoolId);
}
safeShutdownNow(newExecutor); safeShutdownNow(newExecutor);
return false; return false;
} }
ThreadPoolExecutorHolder oldHolder = switchRegistry(threadPoolId, newExecutor);
oldExecutor.shutdown(); oldExecutor.shutdown();
awaitQuietly(oldExecutor, 500, TimeUnit.MILLISECONDS); awaitQuietly(oldExecutor, 500, TimeUnit.MILLISECONDS);
if (!oldExecutor.isTerminated()) { if (!oldExecutor.isTerminated()) {
@ -136,7 +136,14 @@ public final class ThreadPoolRebuilder {
transferredCount++; transferredCount++;
} catch (Throwable ex) { } catch (Throwable ex) {
failedCount++; failedCount++;
log.error("Failed to transfer task during queue switch, may cause task loss.", ex); 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) { } catch (Throwable ex) {
@ -149,11 +156,18 @@ public final class ThreadPoolRebuilder {
transferredCount++; transferredCount++;
} catch (Throwable ex) { } catch (Throwable ex) {
failedCount++; failedCount++;
log.error("Failed to transfer remaining task during queue switch.", ex); 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) { if (transferredCount > 0 || failedCount > 0) {
log.info("Task transfer completed: {} tasks transferred, {} tasks failed.", transferredCount, failedCount); log.info("Task transfer completed: {} tasks transferred, {} tasks failed (restored to old executor).", transferredCount, failedCount);
} }
if (failedCount > 0) { if (failedCount > 0) {
throw new RuntimeException("Task transfer failed: " + failedCount + " tasks could not be transferred."); throw new RuntimeException("Task transfer failed: " + failedCount + " tasks could not be transferred.");

Loading…
Cancel
Save