package com.alibaba.otter.canal.client.adapter.support;

import com.alibaba.druid.pool.DruidDataSource;
import com.google.common.base.Joiner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;

public abstract class AbstractEtlService {

    protected Logger      logger = LoggerFactory.getLogger(this.getClass());

    private String        type;
    private AdapterConfig config;

    public AbstractEtlService(String type, AdapterConfig config){
        this.type = type;
        this.config = config;
    }

    protected EtlResult importData(String sql, List<String> params) {
        EtlResult etlResult = new EtlResult();
        AtomicLong impCount = new AtomicLong();
        List<String> errMsg = new ArrayList<>();
        if (config == null) {
            logger.warn("{} mapping config is null, etl go end ", type);
            etlResult.setErrorMessage(type + "mapping config is null, etl go end ");
            return etlResult;
        }

        long start = System.currentTimeMillis();
        try {
            DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());

            List<Object> values = new ArrayList<>();
            // 拼接条件
            if (config.getMapping().getEtlCondition() != null && params != null) {
                String etlCondition = config.getMapping().getEtlCondition();
                int size = params.size();
                for (String param : params) {
                    etlCondition = etlCondition.replace("{}", "?");
                    values.add(param);
                }

                sql += " " + etlCondition;
            }

            if (logger.isDebugEnabled()) {
                logger.debug("etl sql : {}", sql);
            }

            // 获取总数
            String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
            long cnt = (Long) Util.sqlRS(dataSource, countSql, values, rs -> {
                Long count = null;
                try {
                    if (rs.next()) {
                        count = ((Number) rs.getObject(1)).longValue();
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
                return count == null ? 0L : count;
            });

            // 当大于1万条记录时开启多线程
            if (cnt >= 10000) {
                int threadCount = 3; // 从配置读取默认为3
                long perThreadCnt = cnt / threadCount;
                ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
                List<Future<Boolean>> futures = new ArrayList<>(threadCount);
                for (int i = 0; i < threadCount; i++) {
                    long offset = i * perThreadCnt;
                    Long size = null;
                    if (i != threadCount - 1) {
                        size = perThreadCnt;
                    }
                    String sqlFinal;
                    if (size != null) {
                        sqlFinal = sql + " LIMIT " + offset + "," + size;
                    } else {
                        sqlFinal = sql + " LIMIT " + offset + "," + cnt;
                    }
                    Future<Boolean> future = executor.submit(
                        () -> executeSqlImport(dataSource, sqlFinal, values, config.getMapping(), impCount, errMsg));
                    futures.add(future);
                }

                for (Future<Boolean> future : futures) {
                    future.get();
                }
                executor.shutdown();
            } else {
                executeSqlImport(dataSource, sql, values, config.getMapping(), impCount, errMsg);
            }

            logger.info("数据全量导入完成, 一共导入 {} 条数据, 耗时: {}", impCount.get(), System.currentTimeMillis() - start);
            etlResult.setResultMessage("导入" + type + " 数据：" + impCount.get() + " 条");
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            errMsg.add(type + " 数据导入异常 =>" + e.getMessage());
        }
        if (errMsg.isEmpty()) {
            etlResult.setSucceeded(true);
        } else {
            etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
        }
        return etlResult;
    }

    protected abstract boolean executeSqlImport(DataSource ds, String sql, List<Object> values,
                                                AdapterConfig.AdapterMapping mapping, AtomicLong impCount,
                                                List<String> errMsg);

}

@@ -589,16 +589,62 @@
  config;
+%0A    private final long CNT_PER_TASK = 10000L;
 %0A%0A    pu
@@ -2631,167 +2631,53 @@
 t = 
-3; // %E4%BB%8E%E9%85%8D%E7%BD%AE%E8%AF%BB%E5%8F%96%E9%BB%98%E8%AE%A4%E4%B8%BA3%0A                long perThreadCnt = cnt / threadCount;%0A                ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L
+Runtime.getRuntime().availableProcessors(
 );%0A
+%0A
     
@@ -2692,307 +2692,287 @@
     
-List%3CFuture%3CBoolean%3E%3E futures = new ArrayList%3C%3E(threadCount);%0A                for (int i = 0; i %3C threadCount; i++) %7B%0A                    long offset = i * perThreadCnt;%0A                    Long size = null;%0A                    if (i != threadCount - 1) %7B%0A                        size = perT
+long offset;%0A                long size = CNT_PER_TASK;%0A                long workerCnt = cnt / size + (cnt %25 size == 0 ? 0 : 1);%0A%0A                if (logger.isDebugEnabled()) %7B%0A                    logger.debug(%22workerCnt %7B%7D for cnt %7B%7D threadCount %7B%7D%22, workerCnt, cnt, t
 hreadC
+ou
 nt
+)
 ;%0A  
@@ -2977,37 +2977,34 @@
                 
-   
- 
 %7D
+%0A
 %0A               
@@ -3008,142 +3008,201 @@
     
-    String sqlFinal;%0A                    if (size != null) %7B%0A                        sqlFinal = sql + %22 LIMIT %22 + offset + %22,%22 + size;
+ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);%0A                List%3CFuture%3CBoolean%3E%3E futures = new ArrayList%3C%3E();%0A                for (long i = 0; i %3C workerCnt; i++) %7B
 %0A   
@@ -3214,34 +3214,43 @@
             
-%7D else %7B%0A 
+offset = size * i;%0A
             
@@ -3253,26 +3253,30 @@
             
-  
+String
  sqlFinal = 
@@ -3312,34 +3312,13 @@
 %22 + 
-cnt;%0A                    %7D
+size;
 %0A   

improve es etl (#1758)