canal client leader

//选举

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);  

        CuratorFramework client = CuratorFrameworkFactory.builder().connectString(zookeeper).sessionTimeoutMs(5000).connectionTimeoutMs(3000).retryPolicy(retryPolicy).namespace("canalClient").build();  

        client.start();  

        

        IszLogger.error("=====");

        CanalConnector connector = getCanalConnector(connectorType);

connector.connect();

connector.subscribe(subscribeReg);

IszLogger.error("*****");

 

        // 选举Leader 启动  

        @SuppressWarnings("resource")

final LeaderLatch latch = new LeaderLatch(client,"/canalClient");

          latch.addListener(new LeaderLatchListener() {        

             public void isLeader() {

             //启动监听

             IszLogger.error("leader==============");

          subscribeAndDealCanal();

             }              

             public void notLeader() {

             IszLogger.error("not leader==============");

             }

         });          

         try {

         latch.start(); 

         IszLogger.error("start阻塞==============");

latch.await();

IszLogger.error("通过阻塞================");

} catch (Exception e) {

e.printStackTrace();

}