mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
Merge pull request #1002 from wallellen/fix-release-threadpool
1. AbstractEventParser 中应该对multiStageCoprocess每次在while最后进行stop而不是rese…
This commit is contained in:
@@ -317,10 +317,10 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
|
|||||||
eventSink.interrupt();
|
eventSink.interrupt();
|
||||||
transactionBuffer.reset();// 重置一下缓冲队列,重新记录数据
|
transactionBuffer.reset();// 重置一下缓冲队列,重新记录数据
|
||||||
binlogParser.reset();// 重新置位
|
binlogParser.reset();// 重新置位
|
||||||
if (multiStageCoprocessor != null) {
|
if (multiStageCoprocessor != null && multiStageCoprocessor.isStart()) {
|
||||||
// 处理 RejectedExecutionException
|
// 处理 RejectedExecutionException
|
||||||
try {
|
try {
|
||||||
multiStageCoprocessor.reset();
|
multiStageCoprocessor.stop();
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
logger.debug("multi processor rejected:", t);
|
logger.debug("multi processor rejected:", t);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -138,6 +138,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
|
|||||||
try {
|
try {
|
||||||
parserExecutor.shutdownNow();
|
parserExecutor.shutdownNow();
|
||||||
while (!parserExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
|
while (!parserExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
|
||||||
|
if (parserExecutor.isShutdown() || parserExecutor.isTerminated()) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
parserExecutor.shutdownNow();
|
parserExecutor.shutdownNow();
|
||||||
}
|
}
|
||||||
} catch (Throwable e) {
|
} catch (Throwable e) {
|
||||||
@@ -147,6 +151,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
|
|||||||
try {
|
try {
|
||||||
stageExecutor.shutdownNow();
|
stageExecutor.shutdownNow();
|
||||||
while (!stageExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
|
while (!stageExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
|
||||||
|
if (stageExecutor.isShutdown() || stageExecutor.isTerminated()) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
stageExecutor.shutdownNow();
|
stageExecutor.shutdownNow();
|
||||||
}
|
}
|
||||||
} catch (Throwable e) {
|
} catch (Throwable e) {
|
||||||
|
|||||||
Reference in New Issue
Block a user