actitve-process/.svn/pristine/86/86a345219a5c2681cee9e00eb570589a7a8f4796.svn-base

package com.chinacreator.process.job;

import java.net.URLEncoder;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.PersistJobDataAfterExecution;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;

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.process.bean.KuaishouPushBean;
import com.chinacreator.process.dao.DictionaryDao;
import com.chinacreator.process.dao.KuaishouDao;
import com.chinacreator.process.dao.VipSmsSpidSendDao;
import com.chinacreator.process.util.JsonUtil;
import com.chinacreator.process.util.SHAUtil;
import com.chinacreator.process.util.URLUtil;
import com.chinacreator.process.util.WriteLogUtil;


/**
 * 手动领取会员短信发送
 * 每10分钟执行一次
 * @author xu.zhou
 * @date 20220407
 */
@PersistJobDataAfterExecution
@DisallowConcurrentExecution
public class VipSmsSpidSendJob {

	private static Logger logger = Logger.getLogger("vipSmsSpidSend");
	
	@Autowired
	private VipSmsSpidSendDao vipSmsSpidSendDao; // = new VipSmsSpidSendDao();
	
	@Autowired
	private DictionaryDao dictionaryDao; // = new DictionaryDao();
	
	public void doProcess() throws Exception {
		//System.out.println("VipSmsSpidSendJob定时任务开始"+new SimpleDateFormat("yyyyMMddHHmmss").format(new Date()));
		WriteLogUtil.writeLong("VipSmsSpidSendJob定时任务开始", logger, "VipSmsSpidSendJob");
		int count = 0;
		long beginTime = System.currentTimeMillis();
		//当前月份
		String currmonth = new SimpleDateFormat("yyyyMM").format(new Date());
		int rows = 800; //每次取数据条数
		//获取字典表的配置
		String confrows = dictionaryDao.getValue("vipSmsSpidSendrows");
		if (!StringUtils.isEmpty(confrows)) {
			try {
				rows = Integer.parseInt(confrows);
			} catch (Exception e) {
				rows = 800;
			}
		}
		//采集表分区标识，分区标识从T_HASH_P01到T_HASH_P50
		String partition = "T_HASH_P";
		for(int i = 1; i <= 50; i++){
			count = 0;
			partition = "T_HASH_P";
			try {
				if(i < 10){
					partition = partition + "0" + i;
				}else{
					partition = partition + i;
				}
				
				//按分区标识获取订购数据
				List<HashMap> dataList = vipSmsSpidSendDao.getDataByPart(partition,rows, currmonth);
				count = (dataList != null ? dataList.size() : 0);
				logger.info(partition+",用户数："+(dataList != null ? dataList.size() : "0"));
				if(dataList != null && dataList.size() > 0){					
					CountDownLatch threadSignal = new CountDownLatch(dataList.size());
					ExecutorService executorService = new ThreadPoolExecutor(10, 20, 10L, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(), new ThreadPoolExecutor.CallerRunsPolicy());
					//把数据更新为正在处理状态
					for(HashMap hm : dataList){
						boolean res = vipSmsSpidSendDao.updExecing(hm.get("ID").toString());
					}
					for(HashMap hm : dataList){
						VipSmsSpidSendService continueService = new VipSmsSpidSendService(dataList.size(),threadSignal,hm,currmonth,vipSmsSpidSendDao,dictionaryDao);
				        executorService.execute(continueService);
				    }
					 executorService.shutdown();
				     try {
				          executorService.awaitTermination(5L, TimeUnit.MINUTES);
				     } catch (InterruptedException e) {
				          e.printStackTrace();
				     }
				}
			} catch (Exception e) {
				logger.info(partition+"，执行出现异常，"+e.getMessage());
				e.printStackTrace();
			}
			
			Thread.sleep(100);
		}

		WriteLogUtil.writeLong("VipSmsSpidSendJob定时任务完成，耗时："+(System.currentTimeMillis()-beginTime)/1000+" 秒", logger, "VipSmsSpidSendJob");
	}
	
	public static void main(String[] args) {
		VipSmsSpidSendJob job = new VipSmsSpidSendJob();
		try {
			job.doProcess();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

class VipSmsSpidSendService implements Runnable {

	private static Logger logger = Logger.getLogger("vipSmsSpidSend");
	private int totalSize;
	private CountDownLatch threadSignal;
	private List<String> spidsList;
	private String currmonth;
	private VipSmsSpidSendDao vipSmsSpidSendDao;
	private HashMap hm; 
	private DictionaryDao dictionaryDao;
	
	public VipSmsSpidSendService(int _totalSize,CountDownLatch _threadSignal,HashMap _hm,String _currmonth,
			VipSmsSpidSendDao _vipSmsSpidSendDao, DictionaryDao _dictionaryDao){
		this.totalSize = _totalSize;
		this.threadSignal = _threadSignal;
		this.hm = _hm;
		this.currmonth = _currmonth;
		this.vipSmsSpidSendDao = _vipSmsSpidSendDao;
		this.dictionaryDao = _dictionaryDao;
	}
	
	@Override
	/**
	 * 业务处理
	 * 1. 查询是否配置
	 * 2. 判断当前时间是否在指定时间
	 * 2. 查询当月是否已发送过短信
	 * 3. 发送短信
	 * 4. 更新处理结果
	 */
	public void run() {
		long startime = System.currentTimeMillis();
		Map logMap = new HashMap();
		logMap.put("data", hm);
		String resultcode = "-1";
		String errorinfo = "";
		String id = hm.get("ID").toString();
		try {
			String userid = hm.get("USERID").toString();
			String confid = hm.get("CONFID").toString();
			//获取所有配置信息
			HashMap dataconf = vipSmsSpidSendDao.getVipsmsspidconf(confid);
			//System.out.println("配置数据"+dataconf);
			if(dataconf == null || dataconf.size() == 0){
				throw new BusinessException("9057","无发送配置");
			}
			String smstxt = (String)dataconf.get("SMSTXT");
			if(!StringUtils.isEmpty(smstxt) && smstxt.trim().length()>0){
				//替换手机号码
				smstxt = smstxt.replaceAll("#USERID#", userid);
			}else{
				throw new BusinessException("9053","无短信内容配置");
			}
			//发送时间验证
			String dateformat = (String)dataconf.get("DATEFORMAT");
			String sendstarttime = (String)dataconf.get("SENDSTARTTIME");
			String sendendtime = (String)dataconf.get("SENDENDTIME");
			if(!StringUtils.isEmpty(dateformat)){
				String currtime = new SimpleDateFormat(dateformat).format(new Date());
				logMap.put("currtime", currtime);
				//有配置短信发送开始时间，但当前时间小于开始时间
				if(!StringUtils.isEmpty(sendstarttime) && Long.parseLong(currtime) < Long.parseLong(sendstarttime)){
					throw new BusinessException("9055","未到短信发送开始时间");
				}
				//有配置短信发送结束时间，但当前时间大于结束时间
				if(!StringUtils.isEmpty(sendendtime) && Long.parseLong(currtime) > Long.parseLong(sendendtime)){
					throw new BusinessException("9054","已过短信发送结束时间");
				}
			}
			if(vipSmsSpidSendDao.hasAlreadySend(userid,confid,currmonth)){
				throw new BusinessException("9056","重复发送短信");
			}
			
			HashMap<String, String> params = new HashMap<String, String>();
			params.put("USERID", userid);
			params.put("CONFID", confid);
			params.put("SENDMONTH", currmonth);
			params.put("COLLECTID", id);
			
			//添加发送记录，有唯一约束，添加成功才发短信
			vipSmsSpidSendDao.addSendLog(params);
						
			//发送短信
			String result = send(userid,smstxt);
			errorinfo = result;
			if(!StringUtils.isEmpty(result)){
				JSONObject obj = JSON.parseObject(result);
				resultcode = obj.getString("resultcode");
			}else{
				resultcode = "5000";
				errorinfo = "调发送短信接口无返回信息";
			}
		} catch (Exception e) {
			if (e instanceof BusinessException) {
				errorinfo = ((BusinessException) e).getMessage();
				resultcode = ((BusinessException) e).getCode();
			}else{
				e.printStackTrace();
                if (e.getMessage() != null && e.getMessage().contains("ORA-00001")) {
    				resultcode = "6002";
    				errorinfo += "当月已发送";
                }else{
    				resultcode = "8000";
    				errorinfo += "处理数据出现异常，"+e.getMessage();
                }
			}
		}finally {
			threadSignal.countDown();
			boolean updres = false;
			try{
				if(!StringUtils.isEmpty(errorinfo) && errorinfo.length()>300){
					errorinfo = errorinfo.substring(0,300);
				}
				if("-1".equals(resultcode)){
					resultcode = "8001";
					errorinfo = "处理异常";
				}
				//更新处理结果
				updres = vipSmsSpidSendDao.updExecRes(id,resultcode,errorinfo);				
			}catch(Exception e){
				e.printStackTrace();
				errorinfo += "更新数据出现异常，"+e.getMessage();
				resultcode = "8001";
			}
			//写日志
			logMap.put("jobname", "VipSmsSpidSendJob");
			logMap.put("resultcode", resultcode);
			logMap.put("errorinfo", errorinfo);
			logMap.put("updres", updres);
			logger.info(JsonUtil.objectToJson(logMap));
		}
	}
	
	//发送短信
	public String send(String userid,String content) throws Exception {
		String url = dictionaryDao.getValue("contractpushurl");
//        String url = "http://111.206.133.54/smsbusi/sms/send";
		String smsid = "10655117";
		String pwd = "wo6bslq2";
		userid = DESUtil.encode(userid,pwd);
		String timestamp = System.currentTimeMillis()/1000+"";
		String sign = SHAUtil.shaEncode(smsid+userid+timestamp+content+pwd).toLowerCase();
		userid = URLEncoder.encode(userid,"utf-8");
		content = URLEncoder.encode(content,"utf-8");
		url = url+"?smsid="+smsid+"&userid="+userid+"&timestamp="+timestamp+"&sign="+sign+"&content="+content;
		String result = URLUtil.get(url);

		return result;
	}
}