docs: @Override
      public void init(OuterAdapterConfig configuration, Properties envProperties) {
          this.envProperties = envProperties;
          Map<String, MappingConfig> rdbMappingTmp = ConfigLoader.load(envProperties);
          // 过滤不匹配的key的配置
          rdbMappingTmp.forEach((key, mappingConfig) -> {
              if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
                  || (mappingConfig.getOuterAdapterKey() != null
                      && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
                  rdbMapping.put(key, mappingConfig);
              }
          });
  
          if (rdbMapping.isEmpty()) {
              throw new RuntimeException("No rdb adapter found for config key: " + configuration.getKey());
          }
  
          for (Map.Entry<String, MappingConfig> entry : rdbMapping.entrySet()) {
              String configName = entry.getKey();
              MappingConfig mappingConfig = entry.getValue();
              if (!mappingConfig.getDbMapping().getMirrorDb()) {
                  String key;
                  if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
                      key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-"
                            + StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_"
                            + mappingConfig.getDbMapping().getDatabase() + "-" + mappingConfig.getDbMapping().getTable();
                  } else {
                      key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_"
                            + mappingConfig.getDbMapping().getDatabase() + "-" + mappingConfig.getDbMapping().getTable();
                  }
                  Map<String, MappingConfig> configMap = mappingConfigCache.computeIfAbsent(key,
                      k1 -> new ConcurrentHashMap<>());
                  configMap.put(configName, mappingConfig);
              } else {
                  // mirrorDB
                  String key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
                               + mappingConfig.getDbMapping().getDatabase();
                  mirrorDbConfigCache.put(key, MirrorDbConfig.create(configName, mappingConfig));
              }
          }
  
          // 初始化连接池
          Map<String, String> properties = configuration.getProperties();
          dataSource = new DruidDataSource();
          dataSource.setDriverClassName(properties.get("jdbc.driverClassName"));
          dataSource.setUrl(properties.get("jdbc.url"));
          dataSource.setUsername(properties.get("jdbc.username"));
          dataSource.setPassword(properties.get("jdbc.password"));
          dataSource.setInitialSize(1);
          dataSource.setMinIdle(1);
          dataSource.setMaxActive(30);
          dataSource.setMaxWait(60000);
          dataSource.setTimeBetweenEvictionRunsMillis(60000);
          dataSource.setMinEvictableIdleTimeMillis(300000);
          dataSource.setUseUnfairLock(true);
          // List<String> array = new ArrayList<>();
          // array.add("set names utf8mb4;");
          // dataSource.setConnectionInitSqls(array);
  
          try {
              dataSource.init();
          } catch (SQLException e) {
              logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e);
          }
  
          String threads = properties.get("threads");
          // String commitSize = properties.get("commitSize");
  
          boolean skipDupException = BooleanUtils
              .toBoolean(configuration.getProperties().getOrDefault("skipDupException", "true"));
          rdbSyncService = new RdbSyncService(dataSource,
              threads != null ? Integer.valueOf(threads) : null,
              skipDupException);
  
          rdbMirrorDbSyncService = new RdbMirrorDbSyncService(mirrorDbConfigCache,
              dataSource,
              threads != null ? Integer.valueOf(threads) : null,
              rdbSyncService.getColumnsTypeCache(),
              skipDupException);
  
          rdbConfigMonitor = new RdbConfigMonitor();
          rdbConfigMonitor.init(configuration.getKey(), this, envProperties);
      }