content: package com.alibaba.otter.canal.deployer.monitor;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.TimeUnit;
  
  import org.apache.commons.lang.StringUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
  import com.alibaba.otter.canal.common.CanalLifeCycle;
  import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
  import com.alibaba.otter.canal.instance.manager.plain.PlainCanal;
  import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient;
  import com.google.common.collect.Lists;
  import com.google.common.collect.MapMaker;
  import com.google.common.collect.MigrateMap;
  
  /**
   * 基于manager配置的实现
   *
   * @author agapple 2019年8月26日 下午10:00:20
   * @since 1.1.4
   */
  public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle implements InstanceConfigMonitor, CanalLifeCycle {
  
      private static final Logger         logger               = LoggerFactory.getLogger(ManagerInstanceConfigMonitor.class);
      private long                        scanIntervalInSecond = 5;
      private InstanceAction              defaultAction        = null;
      private Map<String, InstanceAction> actions              = new MapMaker().makeMap();
      private Map<String, PlainCanal>     configs              = MigrateMap.makeComputingMap(destination -> new PlainCanal());
      private ScheduledExecutorService    executor             = Executors.newScheduledThreadPool(1,
                                                                   new NamedThreadFactory("canal-instance-scan"));
  
      private volatile boolean            isFirst              = true;
      private PlainCanalConfigClient      configClient;
  
      public void start() {
          super.start();
          executor.scheduleWithFixedDelay(() -> {
              try {
                  scan();
                  if (isFirst) {
                      isFirst = false;
                  }
              } catch (Throwable e) {
                  logger.error("scan failed", e);
              }
          }, 0, scanIntervalInSecond, TimeUnit.SECONDS);
      }
  
      public void stop() {
          super.stop();
          executor.shutdownNow();
          actions.clear();
      }
  
      public void register(String destination, InstanceAction action) {
          if (action != null) {
              actions.put(destination, action);
          } else {
              actions.put(destination, defaultAction);
          }
      }
  
      public void unregister(String destination) {
          actions.remove(destination);
      }
  
      private void scan() {
          String instances = configClient.findInstances(null);
          if (instances == null) {
              return;
          }
  
          final List<String> is = Lists.newArrayList(StringUtils.split(instances, ','));
          List<String> start = new ArrayList<>();
          List<String> stop = new ArrayList<>();
          List<String> restart = new ArrayList<>();
          for (String instance : is) {
              if (!configs.containsKey(instance)) {
                  PlainCanal newPlainCanal = configClient.findInstance(instance, null);
                  if (newPlainCanal != null) {
                      configs.put(instance, newPlainCanal);
                      start.add(instance);
                  }
              } else {
                  PlainCanal plainCanal = configs.get(instance);
                  PlainCanal newPlainCanal = configClient.findInstance(instance, plainCanal.getMd5());
                  if (newPlainCanal != null) {
                      // 配置有变化
                      restart.add(instance);
                      configs.put(instance, newPlainCanal);
                  }
              }
          }
  
          configs.forEach((instance, plainCanal) -> {
              if (!is.contains(instance)) {
                  stop.add(instance);
              }
          });
  
          stop.forEach(instance -> {
              notifyStop(instance);
          });
  
          restart.forEach(instance -> {
              notifyReload(instance);
          });
  
          start.forEach(instance -> {
              notifyStart(instance);
          });
  
      }
  
      private void notifyStart(String destination) {
          try {
              defaultAction.start(destination);
              actions.put(destination, defaultAction);
              // 启动成功后记录配置文件信息
          } catch (Throwable e) {
              logger.error(String.format("scan add found[%s] but start failed", destination), e);
          }
      }
  
      private void notifyStop(String destination) {
          InstanceAction action = actions.remove(destination);
          if (action != null) {
              try {
                  action.stop(destination);
                  configs.remove(destination);
              } catch (Throwable e) {
                  logger.error(String.format("scan delete found[%s] but stop failed", destination), e);
                  actions.put(destination, action);// 再重新加回去，下一次scan时再执行删除
              }
          }
      }
  
      private void notifyReload(String destination) {
          InstanceAction action = actions.get(destination);
          if (action != null) {
              try {
                  action.reload(destination);
              } catch (Throwable e) {
                  logger.error(String.format("scan reload found[%s] but reload failed", destination), e);
              }
          }
      }
  
      public void release(String destination) {
          InstanceAction action = actions.remove(destination);
          if (action != null) {
              try {
                  configs.remove(destination);
              } catch (Throwable e) {
                  logger.error(String.format("scan delete found[%s] but stop failed", destination), e);
                  actions.put(destination, action);// 再重新加回去，下一次scan时再执行删除
              }
          }
      }
  
      public void setDefaultAction(InstanceAction defaultAction) {
          this.defaultAction = defaultAction;
      }
  
      public void setScanIntervalInSecond(long scanIntervalInSecond) {
          this.scanIntervalInSecond = scanIntervalInSecond;
      }
  
      public void setConfigClient(PlainCanalConfigClient configClient) {
          this.configClient = configClient;
      }
  
      public Map<String, InstanceAction> getActions() {
          return actions;
      }
  
  }
  