IQiYiMain.java

package mobvista.dmp.datasource.iqiyi;

import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
import com.alibaba.fastjson.JSONObject;
import com.google.common.util.concurrent.*;
import mobvista.dmp.common.ClickHouseJdbc;
import mobvista.dmp.util.DateUtil;
import mobvista.dmp.util.PropertyUtil;
import org.apache.commons.lang.StringUtils;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import ru.yandex.clickhouse.ClickHouseConnection;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @package: mobvista.dmp.datasource.iqiyi
 * @author: wangjf
 * @date: 2019-08-31
 * @time: 07:49
 * @email: jinfeng.wang@mobvista.com
 * @phone: 152-1062-7698
 */
public class IQiYiMain {

    static ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5000, 5000, 500, TimeUnit.MILLISECONDS,
            new LinkedBlockingDeque<>(10000), new CustomizableThreadFactory("IQiYi"), new ThreadPoolExecutor.CallerRunsPolicy());

    private static String dt = DateUtil.format(new Date(), "yyyy-MM-dd");

    public static void main(String[] args) throws JoranException, InterruptedException {
        LoggerContext context = (LoggerContext) LoggerFactory.getILoggerFactory();
        JoranConfigurator configurator = new JoranConfigurator();
        configurator.setContext(context);
        context.reset();
        configurator.doConfigure(IQiYiMain.class.getClassLoader().getResourceAsStream("logback-syslog.xml"));
        Logger logger = context.getLogger("IQiYi");

        int fors = 10;
        int concurrent = 2000;
        if (args.length >= 3) {
            fors = Integer.parseInt(args[0]);
            concurrent = Integer.parseInt(args[1]);
            dt = args[2];
        }
        long start = System.currentTimeMillis();

        List<ListenableFuture<IQiYiClass>> futures = new CopyOnWriteArrayList<>();
        long min_start = System.currentTimeMillis();
        for (int c = 1; c <= fors; c++) {

            long in_start = System.currentTimeMillis();

            ListeningExecutorService listeningExecutor = MoreExecutors.listeningDecorator(poolExecutor);
            MoreExecutors.addDelayedShutdownHook(listeningExecutor, 2, TimeUnit.SECONDS);

            ClickHouseConnection connection = null;
            ResultSet resultSet = null;
            try {
                connection = ClickHouseJdbc.connectionByTime(c % 3);
            } catch (SQLException e) {
                try {
                    connection = ClickHouseJdbc.connectionByTime(c % 3);
                } catch (SQLException ex) {
                    logger.info("ClickHouse Connection Failure!");
                }
            }
            String date = DateUtil.getDayByString(dt, "yyyy-MM-dd", -1);
            try {
                resultSet = DMPServer.queryIQiYi(connection, date, concurrent * c, concurrent);
            } catch (SQLException e) {
                logger.info("ClickHouse Query Failure!");
            }
            List<IQiYiClass> resultList = new CopyOnWriteArrayList<>();

            try {
                while (resultSet.next()) {
                    String deviceId = resultSet.getString("device_id");
                    String deviceType = resultSet.getString("device_type");
                    String platform = resultSet.getString("platform");

                    ListenableFuture listenableFuture = listeningExecutor.submit(() -> {
                        JSONObject jsonObject = DMPServer.request(deviceId);
                        IQiYiClass iQiYiClass = new IQiYiClass();
                        iQiYiClass.setDt(dt);
                        iQiYiClass.setDeviceId(deviceId);
                        iQiYiClass.setDeviceType(deviceType);
                        iQiYiClass.setPlatform(platform);
                        if (jsonObject.getInteger("status") == 0) {
                            JSONObject data = jsonObject.getJSONObject("data");
                            for (String key : data.keySet()) {
                                if (data.getIntValue(key) == 1) {
                                    iQiYiClass.setPackageName(getPkgName(platform, key));
                                }
                            }
                            if (StringUtils.isBlank(iQiYiClass.getPackageName())) {
                                iQiYiClass.setPackageName(getPkgName(platform));
                            }
                        } else {
                            iQiYiClass.setPackageName(getPkgName(platform));
                        }
                        return iQiYiClass;
                    });

                    Futures.addCallback(listenableFuture, new FutureCallback<IQiYiClass>() {
                        @Override
                        public void onSuccess(IQiYiClass iQiYiClass) {
                            resultList.add(iQiYiClass);
                        }

                        @Override
                        public void onFailure(Throwable t) {
                            IQiYiClass iQiYiClass = new IQiYiClass();
                            iQiYiClass.setDt(dt);
                            iQiYiClass.setDeviceId(deviceId);
                            iQiYiClass.setDeviceType(deviceType);
                            iQiYiClass.setPlatform(platform);
                            iQiYiClass.setPackageName(getPkgName(platform));
                            resultList.add(iQiYiClass);
                        }
                    });
                    futures.add(listenableFuture);
                }
            } catch (SQLException e) {
                logger.info("ClickHouse Execute Failure!");
            }

            long min_end = System.currentTimeMillis();
            logger.info("Times -->> " + c + ", Runtime -->> " + (min_end - in_start));
            if (c % 10 == 0) {
                try {
                    DMPServer.insertIQiYi(connection, futures);
                    logger.info("Times -->> " + c / 10 + ", ClickHouse Insert Success! Size -->> " + futures.size());
                } catch (SQLException e) {
                    logger.info("ClickHouse Insert Failure!");
                } finally {
                    try {
                        resultSet.close();
                        connection.close();
                    } catch (SQLException throwables) {
                        logger.info("ClickHouse Connection Close!");
                    }
                }
                futures = new CopyOnWriteArrayList<>();
                min_end = System.currentTimeMillis();
                logger.info("ClickHouse Insert Success! Times -->> " + c / 10 + ", Runtime -->> " + (min_end - min_start));
                min_start = System.currentTimeMillis();
            }
        }
        poolExecutor.shutdown();
        long end = System.currentTimeMillis();
        logger.info("all runtime -->> " + (end - start));
    }

    public static String getPkgName(String platform, String tag) {
        return PropertyUtil.getProperty("config.properties", "iqiyi.tag." + tag + "." + platform);
    }

    public static String getPkgName(String platform) {
        String packageName;
        if ("ios".equals(platform)) {
            packageName = "0000000000";
        } else {
            packageName = "com.nonepkg.nonepkg";
        }
        return packageName;
    }
}