datasync / yoho-datasync-producer · Commits

Go to a project

Authored by liangyi.chen@yoho.cn
Commit 6ea6e8e19d5d017592f01620f5f454d377cba7f1 1 parent 59381f84

fix

Inline Side-by-side
Showing 1 changed file with 17 additions and 16 deletions
@@ -90,23 +90,24 @@ public class CanalClientHandler implements InitializingBean, ApplicationEventPub @@ -90,23 +90,24 @@ public class CanalClientHandler implements InitializingBean, ApplicationEventPub
90 for (CanalConfig.CanalInstance instance : canalInstancesList) { 90 for (CanalConfig.CanalInstance instance : canalInstancesList) {
91 threadPool.execute(() -> { 91 threadPool.execute(() -> {
92 Thread.currentThread().setName(Thread.currentThread().getName() + "-" + instance.getDestination()); 92 Thread.currentThread().setName(Thread.currentThread().getName() + "-" + instance.getDestination());
93 - initConnector(instance);  
94 - while (!threadPool.isShutdown() && !threadPool.isTerminated()) {  
95 - try { 93 + try {
  94 + initConnector(instance);
  95 + while (!threadPool.isShutdown() && !threadPool.isTerminated()) {
  96 +
96 handleCanalMessage(canalInstConnectors.get(instance.getDestination()), instance); 97 handleCanalMessage(canalInstConnectors.get(instance.getDestination()), instance);
97 - } catch (CanalClientException e) {  
98 - //TODO 事件上报  
99 - logger.error("CanalClientHandler invoke thread error!", e);  
100 - reConnectCanal(instance);  
101 - } catch (Throwable e) {  
102 - // 避免未捕获的异常导致连接丢失  
103 - // TODO 事件上报  
104 - logger.error("CanalClientHandler invoke thread error!", e);  
105 - try {  
106 - Thread.sleep(1000);  
107 - } catch (InterruptedException ie) {  
108 - Thread.currentThread().interrupt();  
109 - } 98 + }
  99 + }catch (CanalClientException e) {
  100 + //TODO 事件上报
  101 + logger.error("CanalClientHandler invoke thread error!", e);
  102 + reConnectCanal(instance);
  103 + }catch (Throwable e) {
  104 + // 避免未捕获的异常导致连接丢失
  105 + // TODO 事件上报
  106 + logger.error("CanalClientHandler invoke thread error!", e);
  107 + try {
  108 + Thread.sleep(1000);
  109 + } catch (InterruptedException ie) {
  110 + Thread.currentThread().interrupt();
110 } 111 }
111 } 112 }
112 }); 113 });