actitve-process/.svn/pristine/f7/f7c8023566fd8897478edf8b3a6ecf65657eb14f.svn-base

package com.chinacreator.process.job;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.chinacreator.common.exception.BusinessException;
import com.chinacreator.common.util.DESUtil;
import com.chinacreator.common.util.URLUtil;
import com.chinacreator.process.bean.PointShopMqBean;
import com.chinacreator.process.bean.PointShopOrderBean;
import com.chinacreator.process.dao.BackBusiVipAsynDao;
import com.chinacreator.process.dao.DictionaryDao;
import com.chinacreator.process.dao.TaobDao;
import com.chinacreator.process.util.HttpInvoke;
import com.chinacreator.process.util.JsonUtil;
import org.apache.log4j.Logger;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.PersistJobDataAfterExecution;
import org.springframework.beans.factory.annotation.Autowired;

import java.net.URLEncoder;
import java.util.*;
import java.util.concurrent.*;

/**
 * 淘宝业务异步处理（需要回调）
 *
 * @author zhengrong.yan
 * @date 20210127
 */
@PersistJobDataAfterExecution
@DisallowConcurrentExecution
public class TaobSyncJob {

    private Logger log = Logger.getLogger("taobsync");

    @Autowired
    private DictionaryDao dictionaryDao;

    @Autowired
    private TaobDao taobDao;

    @Autowired
    private BackBusiVipAsynDao asyndao;

    public void doProcess() throws Exception {
        log.info(Thread.currentThread().getName() + "定时任务开始");
        long beginTime = System.currentTimeMillis();
        List<HashMap> list = taobDao.getAsynData();
        if (list != null && list.size() > 0) {
            CountDownLatch threadSignal = new CountDownLatch(list.size());
            ExecutorService executorService = new ThreadPoolExecutor(12, 16, 10L, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(), new ThreadPoolExecutor.CallerRunsPolicy());
            //去除重复数据
            List<HashMap> dataList = paraseData(list);
            log.info("数据库有效需要处理的数：" + dataList.size());

            for (HashMap vipmap : dataList) {
                TaobSynService continueService = new TaobSynService(list.size(), threadSignal, vipmap, asyndao, dictionaryDao, taobDao);
                executorService.execute(continueService);
            }
            executorService.shutdown();
            try {
                executorService.awaitTermination(5L, TimeUnit.MINUTES);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        log.info(Thread.currentThread().getName() + "定时任务完成，耗时：" + (System.currentTimeMillis() - beginTime) / 1000 + " 秒");
    }

    /**
     * 去除重复数据，防止赠送会员接口被并发限制
     *
     * @param dataList
     * @return
     */
    private List<HashMap> paraseData(List<HashMap> dataList) {
        //去重复后的数据集
        List<HashMap> reDataList = new ArrayList<HashMap>();
        HashMap<String, List> tmpMap = new HashMap<String, List>();
        for (HashMap dataMap : dataList) {
            if (tmpMap.containsKey(dataMap.get("USERID").toString())) {
                log.info("重复数据，" + dataMap.get("USERID"));
            } else {
                reDataList.add(dataMap);
                List tmpList = new ArrayList();
                tmpList.add(dataMap.get("SPID"));
                tmpMap.put(dataMap.get("USERID").toString(), tmpList);
            }
        }
        return reDataList;
    }

    public static void main(String[] args) {
        TaobSyncJob job = new TaobSyncJob();
        try {
            job.doProcess();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

class TaobSynService implements Runnable {

    private static Logger log = Logger.getLogger("taobsync");
    private int totalSize;
    private CountDownLatch threadSignal;
    private HashMap busimap;
    private DictionaryDao dictionaryDao;
    private BackBusiVipAsynDao asyndao;
    private TaobDao taobDao;

    public TaobSynService(int totalSize, CountDownLatch threadSignal, HashMap busimap, BackBusiVipAsynDao asyndao, DictionaryDao dictionaryDao, TaobDao taobDao) {
        this.totalSize = totalSize;
        this.threadSignal = threadSignal;
        this.busimap = busimap;
        this.asyndao = asyndao;
        this.dictionaryDao = dictionaryDao;
        this.taobDao = taobDao;
    }

    /**
     * 业务处理
     *
     * @param
     */
    public void run() {
        long startime = System.currentTimeMillis();
        Map logMap = new HashMap();
        logMap.put("data", busimap);
        PointShopOrderBean psoBean = new PointShopOrderBean();
        String resultCode = "-1";
        String resultInfo = "";
        String backresult = "";
        try {
            //办理结果编码，0成功，1待处理，2处理中，其他为异常
            psoBean.setId(busimap.get("ID").toString());
            psoBean.setUserid(busimap.get("USERID").toString());
            psoBean.setSpid(busimap.get("SPID").toString());
            psoBean.setCpid(busimap.get("CPID").toString());
            psoBean.setOrderchannel(busimap.get("CHANNEL").toString());
            psoBean.setBusiType(busimap.get("TYPE").toString());
            psoBean.setOrderNo(busimap.get("ORDERID").toString());
            psoBean.setResultCode("2");
            psoBean.setResultInfo("处理中");
            taobDao.upStatus(psoBean);

            //更新订购状态，调接口
            String result = invokeVideoif(psoBean);
            JSONObject jsonObject = JSON.parseObject(result);
            resultCode = jsonObject.getString("resultcode");
            resultInfo = jsonObject.getString("resultinfo");
            if (jsonObject.get("resultinfo") == null) {
                resultInfo = jsonObject.getString("errorinfo");
            }
        } catch (Exception e) {
            e.printStackTrace();
            if (e instanceof BusinessException) {
                resultInfo = ((BusinessException) e).getMessage();
                resultCode = ((BusinessException) e).getCode();
            } else {
                resultCode = "8000";
                resultInfo = "系统错误，" + e.getMessage();
            }
        } finally {
            threadSignal.countDown();
            //更新订购操作表信息，
            try {
                if (resultInfo != null && resultInfo.length() > 250) {
                    resultInfo = resultInfo.substring(0, 250);
                }
                if (psoBean != null) {
                    psoBean.setResultCode(resultCode);
                    psoBean.setResultInfo(resultInfo);
                    //更新后向产品订购记录表处理结果
                    taobDao.updSendStatus(psoBean.getResultCode(), psoBean.getResultInfo(), psoBean.getId());
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.error("更新数据出现异常，" + psoBean.getUserid() + ", resultCode:" + resultCode + ", resultInfo:" + resultInfo);
            }
            if ("0".equals(resultCode)) {
                inserSmstMq(psoBean); //发送短信
            }
            try {
                backresult = this.callBack(psoBean.getUserid(), psoBean.getCpid(), psoBean.getSpid(), psoBean.getBusiType(),psoBean.getId(),busimap.get("ORDERID").toString(),resultCode,resultInfo);
            } catch (Exception e2) {
                e2.printStackTrace();
            }

            logMap.put("data2", psoBean);
            logMap.put("reusltCode", resultCode);
            logMap.put("resultInfo", resultInfo);
            logMap.put("backresult", backresult);
            log.info(JsonUtil.objectToJson(logMap));
        }
    }

    /**
     * 拼接转发URL参数
     *
     * @param orderInfo
     * @return
     * @throws
     */
    private String getParams(PointShopOrderBean orderInfo) {
        String params = "";
        try {
            String cpid = orderInfo.getCpid();
            String spid = orderInfo.getSpid();
            String type = orderInfo.getBusiType();
            String userid = orderInfo.getUserid();
            String timestamp = (System.currentTimeMillis() / 1000) + "";
            String seqnumber = "";
            String seqnumberParams = "";
            if (!"".equals(seqnumber)) {
                seqnumberParams = "&seqnumber=" + seqnumber;
            }
            String pwd = taobDao.getPwd(cpid, spid); //TB_CP_ACCOUNT_CONFIG表NETPWD字段
            if ("0".equals(type)) {//订购
                params += "ordertype=0&access=1&apptype=2&channel=" + orderInfo.getOrderchannel() + "&cpid=" + cpid + "&spid=" + spid + "&timestamp=" + timestamp + "&token=";
            } else {//退订
                params += "ordertype=0&access=1&apptype=2&channel=" + orderInfo.getOrderchannel() + "&cpid=" + cpid + "&spid=" + spid + "&timestamp=" + timestamp;
            }

            userid = DESUtil.encode(userid, pwd);//手机号码加密
            params += "&userid=" + URLEncoder.encode(userid, "utf-8"); //重新编码
            params += seqnumberParams;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return params;
    }


    /**
     * 调videoif-order.do接口
     *
     * @param
     * @return
     * @throws Exception
     */
    private String invokeVideoif(PointShopOrderBean orderBean) throws Exception {
        String result = "";
        try {
            String url = dictionaryDao.getValue("commonOrderUrl");
            if (orderBean.getBusiType().equals("0")) {
                url += "order.do?";
            } else {
                url = "cancelOrder.do?";
            }
            String params = getParams(orderBean);
            url = url + params;
            log.info("tburl: " + url);

            result = URLUtil.get(url); //调赠送会员接口，超时时间设置为10秒
            log.info("调videoif结果=> userid: " + orderBean.getUserid() + ", orderid: " + orderBean.getOrderNo() + " , result: " + result);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("id=>" + orderBean.getId() + "=>userid: " + orderBean.getUserid() + "调videoif失败，" + e);
        }

        return result;
    }

    /**
     * 回调淘宝接口
     *
     * @param
     * @param id
     */
    private String callBack(String userid, String cpid, String spid, String status,String id,String orderid,String resultCode,String resultinfo) {
        String bcstatus = "1"; //回调状态，1未回调，0回调完成，2回调异常
        String cbReqParams = ""; //回调请求报文
        String cbRspParams = ""; //回调响应报文
        try {
            PointShopOrderBean bean = taobDao.getOrderRec(userid, cpid, spid);
               TreeMap<String, String> orderOutObj = null;
                orderOutObj = new TreeMap<String, String>();
                orderOutObj.put("orderid", orderid);
                orderOutObj.put("userid", userid);
                orderOutObj.put("cpid", cpid);
                orderOutObj.put("spid", spid);
                orderOutObj.put("resultCode", resultCode);
                orderOutObj.put("resultinfo", resultinfo);
            if (bean != null) {
                String reusltCode = bean.getResultCode(); //办理结果编码，0成功，1待处理，2处理中，其他为异常
                if ("0".equals(reusltCode)) {
                    orderOutObj.put("status", status);//状态 订购0退订1
                    if ("0".equals(status)) {
                        orderOutObj.put("ordertime", bean.getOrderTime());
                    } else {
                        orderOutObj.put("canceltime", bean.getCanceltime());
                    }
                }
                orderOutObj.put("resultcode", reusltCode);
            }
                String callBackUrl = this.dictionaryDao.getValue("taobCallBackUrl");
                cbReqParams = JsonUtil.objectToJson(orderOutObj);
                log.info("===========回调地址：" + callBackUrl);
                log.info("===========回调请求参数：" + cbReqParams);
                cbRspParams = HttpInvoke.sendHttpByPost("POST", callBackUrl, cbReqParams, getProperty());
                log.info("============回调响应参数:" + cbRspParams);
                if ("0000".equals(cbRspParams)) {
                    bcstatus = "0";
                }

        } catch (Exception e) {
            bcstatus = "2";
            e.printStackTrace();
            log.error("订单=>" + userid + "=>cpid=>" + cpid + "=>spid=>" + spid + "，回调接口出现异常，" + e.getMessage());
            cbRspParams = "返回报文: " + cbRspParams + "，回调异常，" + e.getMessage();
        } finally {
            if (cbRspParams != null && cbRspParams.length() > 250) {
                cbRspParams = cbRspParams.substring(0, 250);
            }
            log.info("数据库需处理id==>" + id + "=>cbReqParams: " + cbReqParams + ", cbRspParams=>" + cbRspParams);
            addParamsByCb(cbRspParams, id);
        }
        return bcstatus;
    }

    /**
     * 添加回调报文
     *
     * @param requestId
     */
    private void addParamsByCb(String rspParams, String requestId) {
        try {
            //一个中文占两个 字节
            if (rspParams != null && rspParams.length() > 100) {
                rspParams = rspParams.substring(0, 100);
            }
            taobDao.updParamsByCb(rspParams, requestId);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("requestId: " + requestId + ", 更新callBack状态出现异常，" + e.getMessage());
        }
    }

    /**
     * 解析数据
     *
     * @param body orderId TD_POINTS_ORDER_REC表ID字段，我方生成的订单流水号
     *             orderNo 积分商城订单号
     *             requestId 返回给客户的请求ID
     * @return
     */
    public PointShopMqBean transBean(Map<String, Object> body) {
        String jsonStr = JsonUtil.objectToJson(body);
        return (PointShopMqBean) JsonUtil.jsonToBean(jsonStr, PointShopMqBean.class);
    }

    /**
     * 发送短信
     *
     * @param psoBean
     */
    public void inserSmstMq(PointShopOrderBean psoBean) {
        try {
            if (psoBean != null) {
                Map<String, String> map = new HashMap<String, String>();
                map.put("userid", psoBean.getUserid());
                map.put("cpid", psoBean.getCpid());
                map.put("spid", psoBean.getSpid());
                map.put("result", "0");
                map.put("channel", "tbcjhd");
                map.put("style", "0000");
                map.put("times", "");
                map.put("orderType", "");
                map.put("type", "cssms");
                if (psoBean.getBusiType().equals("0")) {
                    map.put("busiType", "tran_succ_tb");
                } else {
                    map.put("busiType", "cancel_succ_tb");
                }

                log.info(JsonUtil.objectToJson(map));
                String mqReciveUrl = dictionaryDao.getValue("mqReciveUrl");
                URLUtil.post(mqReciveUrl, JsonUtil.objectToJson(map));
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取请求属性性
     *
     * @return
     */
    private static Map getProperty() {
        Map reqProperty = new HashMap();
        reqProperty.put("Content-type", "application/json;charset=UTF-8");
        return reqProperty;
    }

}