docs: synchronized void start(Properties properties) throws Throwable {
          String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE);
          if (serverMode.equalsIgnoreCase("kafka")) {
              canalMQProducer = new CanalKafkaProducer();
          } else if (serverMode.equalsIgnoreCase("rocketmq")) {
              canalMQProducer = new CanalRocketMQProducer();
          }
  
          if (canalMQProducer != null) {
              // disable netty
              System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true");
              String autoScan = CanalController.getProperty(properties, CanalConstants.CANAL_AUTO_SCAN);
              // 设置为raw避免ByteString->Entry的二次解析
              System.setProperty("canal.instance.memory.rawEntry", "false");
              if ("true".equals(autoScan)) {
                  String rootDir = CanalController.getProperty(properties, CanalConstants.CANAL_CONF_DIR);
                  if (StringUtils.isEmpty(rootDir)) {
                      rootDir = "../conf";
                  }
                  File rootdir = new File(rootDir);
                  if (rootdir.exists()) {
                      File[] instanceDirs = rootdir.listFiles(new FileFilter() {
  
                          public boolean accept(File pathname) {
                              String filename = pathname.getName();
                              return pathname.isDirectory() && !"spring".equalsIgnoreCase(filename) &&
                                      !"metrics".equalsIgnoreCase(filename);
                          }
                      });
                      if (instanceDirs != null && instanceDirs.length > 0) {
                          List<String> instances = Lists.transform(Arrays.asList(instanceDirs),
                              new Function<File, String>() {
  
                                  @Override
                                  public String apply(File instanceDir) {
                                      return instanceDir.getName();
                                  }
                              });
                          System.setProperty(CanalConstants.CANAL_DESTINATIONS, Joiner.on(",").join(instances));
                      }
                  }
              } else {
                  String destinations = CanalController.getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
                  System.setProperty(CanalConstants.CANAL_DESTINATIONS, destinations);
              }
          }
  
          logger.info("## start the canal server.");
          controller = new CanalController(properties);
          controller.start();
          logger.info("## the canal server is running now ......");
          shutdownThread = new Thread() {
  
              public void run() {
                  try {
                      logger.info("## stop the canal server");
                      controller.stop();
                      CanalLauncher.running = false;
                  } catch (Throwable e) {
                      logger.warn("##something goes wrong when stopping canal Server:", e);
                  } finally {
                      logger.info("## canal server is down.");
                  }
              }
  
          };
          Runtime.getRuntime().addShutdownHook(shutdownThread);
  
          if (canalMQProducer != null) {
              canalMQStarter = new CanalMQStarter(canalMQProducer);
              MQProperties mqProperties = buildMQProperties(properties);
              canalMQStarter.start(mqProperties);
              controller.setCanalMQStarter(canalMQStarter);
          }
      }