AlipayRequestMain.java

package mobvista.dmp.datasource.ali;

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.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alipay.api.AlipayApiException;
import com.alipay.api.AlipayClient;
import com.alipay.api.DefaultAlipayClient;
import com.alipay.api.request.AlipayUserAccountDeviceInfoQueryRequest;
import com.alipay.api.response.AlipayUserAccountDeviceInfoQueryResponse;
import com.google.common.util.concurrent.*;
import mobvista.dmp.common.ClickHouseJdbc;
import mobvista.dmp.util.DateUtil;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.except.ClickHouseUnknownException;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

import com.alipay.api.internal.util.AlipayLogger;

public class AlipayRequestMain {
    public static Logger logger = null;

    static ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(500, 900, 500, TimeUnit.MILLISECONDS,
            new LinkedBlockingDeque<>(900), new CustomizableThreadFactory("AlipayRequestMain-"), 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(UCRequest.class.getClassLoader().getResourceAsStream("logback-syslog.xml"));
        logger = context.getLogger("AlipayRequestMain");

        int fors = 30;
        int concurrent = 10000;
        String device_type="";
        String hour="";
        logger.info("args[3]====="+args[3]);
        logger.info("args[3].length()====="+args[3].length());
        if (args.length >= 5) {
            fors = Integer.parseInt(args[0]);
            concurrent = Integer.parseInt(args[1]);
            dt = args[2];
            device_type = args[3];
            hour = args[4];
        }
        logger.info("device_type====="+device_type);
        logger.info("hour====="+hour);

        long start = System.currentTimeMillis();
        List<ListenableFuture<String>> futures = new CopyOnWriteArrayList<>();

        long min_start = System.currentTimeMillis();
        for (int c = 1; c <= fors; c++) {
            int indexNum=1;
            long in_start = System.currentTimeMillis();
            ListeningExecutorService listeningExecutor = MoreExecutors.listeningDecorator(poolExecutor);
            MoreExecutors.addDelayedShutdownHook(listeningExecutor, 5, TimeUnit.SECONDS);
            ClickHouseConnection connection = null;
            ResultSet resultSet = null;
            try {
                connection = ClickHouseJdbc.connectionByTime(c);
            } catch (SQLException e) {
                try {
                    connection = ClickHouseJdbc.connectionByTime(c);
                } catch (SQLException ex) {
                    logger.info("ClickHouse Connection Failure!");
//                    System.out.println("ClickHouse Connection Failure!");
                }
            }
            String date = DateUtil.getDayByString(dt, "yyyy-MM-dd", 0);
            try {
                assert connection != null;
                resultSet = query(connection, date, hour, device_type,concurrent * (c - 1), concurrent);
            } catch (SQLException e) {
                logger.info("ClickHouse Query Failure!");
//                System.out.println("ClickHouse Query Failure!");
            }
            List<String> resultList = new CopyOnWriteArrayList<>();
            try {
                while (resultSet.next()) {
                    indexNum++;
                    String device_ids = resultSet.getString("device_ids");
                    String finalDevice_type = device_type.substring(0,4);
                    int finalIndexNum = indexNum;
                    ListenableFuture listenableFuture = listeningExecutor.submit(() -> {
                        String[] split = device_ids.split(",");
                        List<String> deviceIds = Arrays.asList(split);
                        if(finalIndexNum %1000==0) {
                            return doGetJsonByAliCodeWithExceptionInfo(finalDevice_type, deviceIds);
                        }else{
                            return doGetJsonByAliCode(finalDevice_type, deviceIds);
                        }
                    });

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

                        @Override
                        public void onFailure(Throwable t) {
                            resultList.add(new JSONObject().toJSONString());
                        }
                    });
                    futures.add(listenableFuture);
                }
            } catch (SQLException e) {
                logger.info("ClickHouse Execute Failure!");
//                System.out.println("ClickHouse Execute Failure!");
            }
            long min_end = System.currentTimeMillis();
            logger.info("Times -->> " + c + ", Runtime -->> " + (min_end - in_start));
//            System.out.println("Times -->> " + c + ", Runtime -->> " + (min_end - in_start));
            if (c % 10 == 0) {
                try {
//                    logger.info("Times -->> " + c / 20 + ", ClickHouse begin Insert~~");
                    System.out.println("Times -->> " + c / 10 + ", ClickHouse begin Insert~~");
//                    String datetime = DateUtil.format(new Date(), "yyyy-MM-dd HH");
                    insertAlipay(connection, futures, date, hour, device_type);
                    logger.info("Times -->> " + c / 10 + ", ClickHouse Insert Success! Size -->> " + futures.size());
//                    System.out.println("Times -->> " + c / 20 + ", ClickHouse Insert Success! Size -->> " + futures.size());
                } catch (SQLException e) {
                    logger.info("ClickHouse Insert Failure!");
//                    System.out.println("ClickHouse Insert Failure!");
                }
                futures = new CopyOnWriteArrayList<>();
                min_end = System.currentTimeMillis();
                logger.info("ClickHouse Insert Success! Times -->> " + c / 10 + ", Runtime -->> " + (min_end - min_start));
//                System.out.println("ClickHouse Insert Success! Times -->> " + c / 20 + ", Runtime -->> " + (min_end - min_start));
                min_start = System.currentTimeMillis();
            }
        }
        poolExecutor.shutdown();
        long end = System.currentTimeMillis();
        logger.info("all runtime -->> " + (end - start));
//        System.out.println("all runtime -->> " + (end - start));
    }


    public static ResultSet query(Connection connection, String dt,  String hour ,String device_type, int offSet, int size) throws SQLException, InterruptedException {
        ResultSet resultSet;
        String sql = "SELECT device_ids FROM dmp.alipay_lahuo_daily_all WHERE dt = '@dt' and hour = '@hour' and device_type = '@device_type' LIMIT @offSet, @size";
        sql = sql.replace("@dt", dt).replace("@hour", hour).replace("@device_type", device_type).replace("@offSet", String.valueOf(offSet)).replace("@size", String.valueOf(size));
        try {
            resultSet = connection.createStatement().executeQuery(sql);
            return resultSet;
        } catch (ClickHouseUnknownException e) {
            Thread.sleep(100);
            resultSet = connection.createStatement().executeQuery(sql);
            return resultSet;
        }
    }


    public static void insertAlipay(Connection connection, List<ListenableFuture<String>> futures, String dt, String hour,String device_type) throws SQLException, InterruptedException {
        PreparedStatement preparedStatement = null;
        try {
            preparedStatement = connection.prepareStatement("INSERT INTO dmp.alipay_lahuo_result_daily (dt, hour, device_type,device_ids) VALUES (?, ?, ?, ?)");
//            logger.info("INSERT begin===");
            for (ListenableFuture<String> future : futures) {
                try {
                    String result = future.get();
                    preparedStatement.setString(1, dt);
                    preparedStatement.setString(2, hour);
                    preparedStatement.setString(3, device_type);
                    preparedStatement.setString(4, result);
                    preparedStatement.addBatch();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
            preparedStatement.executeBatch();
//            logger.info("INSERT SUCCESS!");
        } catch (ClickHouseUnknownException | NullPointerException e) {
            logger.info("INSERT -->> " + e.getMessage());
//            System.out.println("INSERT -->> " + e.getMessage());
            Thread.sleep(100);
            assert preparedStatement != null;
            preparedStatement.executeBatch();
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
    }


    public static JSONObject doGetJsonByURI(URI url) throws IOException {
        JSONObject jsonObject = null;
        CloseableHttpClient client = HttpClients.createDefault();
        HttpGet httpGet = new HttpGet(url);
        HttpResponse response = client.execute(httpGet);
        HttpEntity entity = response.getEntity();
        if (entity != null) {
            String result = EntityUtils.toString(entity, "UTF-8");
            jsonObject = JSON.parseObject(result);
        }
        return jsonObject;
    }


    public static String doGetJsonByAliCode(String deviceType, List<String> deviceIds) throws AlipayApiException {
        JSONObject biz_content = new JSONObject();
        biz_content.put("device_type",deviceType);
        biz_content.put("device_ids",deviceIds);
        biz_content.put("encrypt_type","MD5");
        biz_content.put("request_from","M0006");

        String APP_ID="2021002110697881";
        String APP_PRIVATE_KEY="MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCBtpvxY6oSW97jJXviMmaaYIPsqt3FX/mzI5iNB97dZM0+OOtiul0yDcVbKunjKOJYXBvvHc1x2V1GR5xPUrwJSugMuLRVyniKTl7Iz3qbZvyfinrHYqxY+Ll3wIs/FzkP+COqjTzcTGPM/GdFuRYiPrBtD+uvLGQ8MxTuzXRh5RWjsqmObnYOgDIO4RkBCFkOv+mymlfurXdlMAjcb2KuwV/JqPIBjFC/iwBWNQxYbiLvUfpJUlsb4ErTpf4q4iFRPgOEGpak6H4rwwMQ1rkeFMn9IVOXzVnk28r8b7o1WoRjeC7k/y1Qr0Z7HBJu/sOb0iUS+SiyrQI67kvO5VGHAgMBAAECggEARuNtRfaZs37RyIFruqqguC2W63hVytAwvl3tuOVcPtIf+KlSupnu5LPx72j8eVCWooN0ILHA3T9VYlswaYllig6N0qX+qAQG7E74LfZ1owgdd3NOtjiOx/DmKzPm5Yvje364vzSA1kRJJnWqWshT+FMaSAAzZYB1JxEjl35g5D5u7te6hwaeU+jHWa7+StxB870sV7HYRaIawPr4zcBS/MUkS6VbmEk4DjSfa+sULCfrdEP7g/4TCo8xvx71DWGLDeDXNQpTa9ff/zSd0O9XLionfvY4sQiL6W1WTDz7oGOdK7reg2/V/UMeGG9fh8TlE1Uo6M0MnWdOeC9OIHmwgQKBgQC3TNACpa4F/xVY4caUc/yLgKdPTRO/DCx7wbFjkB92BykKAVmgIIK5SJTlYOgqCBRu9/FGi8yNSSTkwvkQ7FSAmor9R4WW5sab4/ZcpImNgrij/hcbD4mO7ZC4tMKPGsKYm5120dxlTbBIa3x4p5lqTHwTJf4Odn7TdsW4jjlvXwKBgQC1KOvZLilTriib+Ym8O24tZARcQ1gqlgkg5hHf2K8RjqnEiWYVV0+UNaCFp4Qe0SZ4sl03MzCrDXDrBfpDVAGWiiCasa2VZhtlHhTVgCnjhO92umG14PGlNlrpnOj7FysCJKi6B/CuL8LQWQ/nm8rPjxPESczrKE992fXeGrRW2QKBgQCgSvTQi9BeAA7L9kyw/6kG2Y/2QkIVXPD9nImitqM8lNBnpnyZEjUfEIDdAlykfQJZdEuzVIehuiggQPyLtw/CkuMe0M/bAZ8HLlf/9gcwD/OlvYq5KGIz8j4PuEs317/6soMGKEmi8UhKXGyyGmaH1bDfEedVAHyyMfiAByfk4wKBgChbnW8coYhK3DEjPlP4yWrDRVVoi1OxaQczOfgsRZQbvCZUlbFbahcSe1hLIVvcLD3nPmz1snno3Kb/BenJ5TCsROcxfgIyZRf5w5psmuhczRweGXZzTmIH4ej5U83cM3CeyhfnfO7RJQ1VSf3G7z+eGEE6T7RznoHi9KsuN87BAoGAKPpzm5DDgNRPEe7dTCV3g8OS2SjuQEavKGR53GM76z2MC9o7GR8UNXUz8OPRz/1V9/3Qzv4mQZfXdu4ULSnuQQaGcV4TL2eATWRty0IsomTkLeHkNJGIqi/F8LxXxDh7fr0D3qcbjv7KZl96axPrFto+Tk1UyC+00ox/4Rv3m3c=";
        String CHARSET="utf-8";
        String ALIPAY_PUBLIC_KEY="MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAjXtI2v/T2/lRwnryxoS2P/yuVCsjvuyCy1exn/LAFkevVSQzxYQt+yaj7FXGCSaIbsdmanFACcTUq7Go5Y09g3AbJELhABckHU7DgUbdSZls48TIXd4tzrBt0OBUDYC1rZ0iqPwomMNREyA7mqXQcGWcR0prSFdMUx7NmZihDOh7D9rAu6fliBAMRLKzC3uqLDJUlPdiJeo/qto1mircWvzxZWaZLgMcYF/w5c9km4f6mIUjpNez43jDlEhsV+/HR+CxesKllpJlDJ3P7XUdqho2GErCrgGGA9uUF3mT0sjYPLpJVs5uc6/wYvbugkVSjbcCYHdfKGD6IbyGtEhQSwIDAQAB";

        String result="";
        try {
            //实例化客户端
            AlipayClient alipayClient = new DefaultAlipayClient("https://ugapi.alipay.com/gateway.do", APP_ID, APP_PRIVATE_KEY, "json", CHARSET, ALIPAY_PUBLIC_KEY, "RSA2");
            //屏蔽支付宝sdk日志
            AlipayLogger.setNeedEnableLogger(false);
            //实例化具体API对应的request类,类名称和接口名称对应,当前调用接口名称：alipay.user.account.device.info.query
            AlipayUserAccountDeviceInfoQueryRequest request = new AlipayUserAccountDeviceInfoQueryRequest();
            //SDK已经封装掉了公共参数，这里只需要传入业务参数
            //此次只是参数展示，未进行字符串转义，实际情况下请转义
            //        request.setBizContent("  {" +
            //                "    \"primary_industry_name\":\"IT科技/IT软件与服务\"," +
            //                "    \"primary_industry_code\":\"10001/20102\"," +
            //                "    \"secondary_industry_code\":\"10001/20102\"," +
            //                "    \"secondary_industry_name\":\"IT科技/IT软件与服务\"" +
            //                " }");
            request.setBizContent(biz_content.toString());
            AlipayUserAccountDeviceInfoQueryResponse response = alipayClient.execute(request);
            //调用成功，则处理业务逻辑
            if (response.isSuccess()) {
                result = response.getBody();
            }
        }catch (Exception e) {
//            支付宝api调用时，如果访问链接数目过大（初步猜测是这个原因），会报异常信息，信息较多，打印出来影响程序速度，所以屏蔽了异常打印语句
//            e.printStackTrace();
//            System.out.println("Exception="+e.getMessage());
        }
        return result;
    }

    public static String doGetJsonByAliCodeWithExceptionInfo(String deviceType, List<String> deviceIds) throws AlipayApiException {
        JSONObject biz_content = new JSONObject();
        biz_content.put("device_type",deviceType);
        biz_content.put("device_ids",deviceIds);
        biz_content.put("encrypt_type","MD5");
        biz_content.put("request_from","M0006");

        String APP_ID="2021002110697881";
        String APP_PRIVATE_KEY="MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCBtpvxY6oSW97jJXviMmaaYIPsqt3FX/mzI5iNB97dZM0+OOtiul0yDcVbKunjKOJYXBvvHc1x2V1GR5xPUrwJSugMuLRVyniKTl7Iz3qbZvyfinrHYqxY+Ll3wIs/FzkP+COqjTzcTGPM/GdFuRYiPrBtD+uvLGQ8MxTuzXRh5RWjsqmObnYOgDIO4RkBCFkOv+mymlfurXdlMAjcb2KuwV/JqPIBjFC/iwBWNQxYbiLvUfpJUlsb4ErTpf4q4iFRPgOEGpak6H4rwwMQ1rkeFMn9IVOXzVnk28r8b7o1WoRjeC7k/y1Qr0Z7HBJu/sOb0iUS+SiyrQI67kvO5VGHAgMBAAECggEARuNtRfaZs37RyIFruqqguC2W63hVytAwvl3tuOVcPtIf+KlSupnu5LPx72j8eVCWooN0ILHA3T9VYlswaYllig6N0qX+qAQG7E74LfZ1owgdd3NOtjiOx/DmKzPm5Yvje364vzSA1kRJJnWqWshT+FMaSAAzZYB1JxEjl35g5D5u7te6hwaeU+jHWa7+StxB870sV7HYRaIawPr4zcBS/MUkS6VbmEk4DjSfa+sULCfrdEP7g/4TCo8xvx71DWGLDeDXNQpTa9ff/zSd0O9XLionfvY4sQiL6W1WTDz7oGOdK7reg2/V/UMeGG9fh8TlE1Uo6M0MnWdOeC9OIHmwgQKBgQC3TNACpa4F/xVY4caUc/yLgKdPTRO/DCx7wbFjkB92BykKAVmgIIK5SJTlYOgqCBRu9/FGi8yNSSTkwvkQ7FSAmor9R4WW5sab4/ZcpImNgrij/hcbD4mO7ZC4tMKPGsKYm5120dxlTbBIa3x4p5lqTHwTJf4Odn7TdsW4jjlvXwKBgQC1KOvZLilTriib+Ym8O24tZARcQ1gqlgkg5hHf2K8RjqnEiWYVV0+UNaCFp4Qe0SZ4sl03MzCrDXDrBfpDVAGWiiCasa2VZhtlHhTVgCnjhO92umG14PGlNlrpnOj7FysCJKi6B/CuL8LQWQ/nm8rPjxPESczrKE992fXeGrRW2QKBgQCgSvTQi9BeAA7L9kyw/6kG2Y/2QkIVXPD9nImitqM8lNBnpnyZEjUfEIDdAlykfQJZdEuzVIehuiggQPyLtw/CkuMe0M/bAZ8HLlf/9gcwD/OlvYq5KGIz8j4PuEs317/6soMGKEmi8UhKXGyyGmaH1bDfEedVAHyyMfiAByfk4wKBgChbnW8coYhK3DEjPlP4yWrDRVVoi1OxaQczOfgsRZQbvCZUlbFbahcSe1hLIVvcLD3nPmz1snno3Kb/BenJ5TCsROcxfgIyZRf5w5psmuhczRweGXZzTmIH4ej5U83cM3CeyhfnfO7RJQ1VSf3G7z+eGEE6T7RznoHi9KsuN87BAoGAKPpzm5DDgNRPEe7dTCV3g8OS2SjuQEavKGR53GM76z2MC9o7GR8UNXUz8OPRz/1V9/3Qzv4mQZfXdu4ULSnuQQaGcV4TL2eATWRty0IsomTkLeHkNJGIqi/F8LxXxDh7fr0D3qcbjv7KZl96axPrFto+Tk1UyC+00ox/4Rv3m3c=";
        String CHARSET="utf-8";
        String ALIPAY_PUBLIC_KEY="MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAjXtI2v/T2/lRwnryxoS2P/yuVCsjvuyCy1exn/LAFkevVSQzxYQt+yaj7FXGCSaIbsdmanFACcTUq7Go5Y09g3AbJELhABckHU7DgUbdSZls48TIXd4tzrBt0OBUDYC1rZ0iqPwomMNREyA7mqXQcGWcR0prSFdMUx7NmZihDOh7D9rAu6fliBAMRLKzC3uqLDJUlPdiJeo/qto1mircWvzxZWaZLgMcYF/w5c9km4f6mIUjpNez43jDlEhsV+/HR+CxesKllpJlDJ3P7XUdqho2GErCrgGGA9uUF3mT0sjYPLpJVs5uc6/wYvbugkVSjbcCYHdfKGD6IbyGtEhQSwIDAQAB";

        String result="";
        try {
            //实例化客户端
            AlipayClient alipayClient = new DefaultAlipayClient("https://ugapi.alipay.com/gateway.do", APP_ID, APP_PRIVATE_KEY, "json", CHARSET, ALIPAY_PUBLIC_KEY, "RSA2");
            //屏蔽支付宝sdk日志
            AlipayLogger.setNeedEnableLogger(false);
            //实例化具体API对应的request类,类名称和接口名称对应,当前调用接口名称：alipay.user.account.device.info.query
            AlipayUserAccountDeviceInfoQueryRequest request = new AlipayUserAccountDeviceInfoQueryRequest();
            //SDK已经封装掉了公共参数，这里只需要传入业务参数
            //此次只是参数展示，未进行字符串转义，实际情况下请转义
            //        request.setBizContent("  {" +
            //                "    \"primary_industry_name\":\"IT科技/IT软件与服务\"," +
            //                "    \"primary_industry_code\":\"10001/20102\"," +
            //                "    \"secondary_industry_code\":\"10001/20102\"," +
            //                "    \"secondary_industry_name\":\"IT科技/IT软件与服务\"" +
            //                " }");
            request.setBizContent(biz_content.toString());
            AlipayUserAccountDeviceInfoQueryResponse response = alipayClient.execute(request);
            //调用成功，则处理业务逻辑
            if (response.isSuccess()) {
                result = response.getBody();
            }
        }catch (Exception e) {
            e.printStackTrace();
            System.out.println("Exp="+e.getMessage());
        }
        return result;
    }
}