iamberry-common-parent/wateroPF-wechat-web/src/main/java/com/iamberry/wechat/handles/mq/MQTask.java

package com.iamberry.wechat.handles.mq;

import java.lang.reflect.Method;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.iamberry.wechat.core.entity.admin.ShopSystemRule;
import com.iamberry.wechat.core.entity.order.Order;
import com.iamberry.wechat.face.admin.SystemService;
import com.iamberry.wechat.tools.HttpClient431Util;
import com.iamberry.wechat.tools.NameUtils;
import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.StringEscapeUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.iamberry.app.tool.log.RatFWLogger;
import com.iamberry.wechat.core.entity.mq.MQMessage;
import com.iamberry.wechat.core.entity.order.ProbationOrderDto;
import com.iamberry.wechat.core.entity.probation.ProbationAwardRelu;
import com.iamberry.wechat.core.entity.probation.ProbationNewLogs;
import com.iamberry.wechat.core.entity.task.TaskModel;
import com.iamberry.wechat.core.entity.task.WechatTask;
import com.iamberry.wechat.core.entity.task.WechatTaskLogs;
import com.iamberry.wechat.face.order.CodeService;
import com.iamberry.wechat.face.order.ProbationShopOrderService;
import com.iamberry.wechat.face.reback.RebackServices;
import com.iamberry.wechat.face.task.WechatTaskService;
import com.iamberry.wechat.service.StaticCacheMemory;
import com.iamberry.wechat.tools.DateTimeUtil;
import com.iamberry.wechat.tools.ResultInfo;
import com.iamberry.wechat.utils.SendMessageUtil;
import com.iamberry.zk.SpringContextHolder;
/**
 * @author:何秀刚
 * @description: MQ模拟实现，每天最多处理1382400条
 * @createDate:2016年5月25日
 */
@Component(value="mqTask")
@Lazy(false)
public class MQTask implements InitializingBean {
	
	@Autowired
	private MQServiceProxy mQSerivce;
	
	@Autowired
	private RatFWLogger logger;
	
	@Autowired
	private WechatTaskService wechatTaskService;
	
	@Autowired
	private RebackServices rebackServices;
	
	@Autowired
	private ProbationShopOrderService probationShopOrderService;
	
	@Autowired
	private SendMessageUtil sendMessageUtil;
	
	@Autowired
	private CodeService codeService;
	@Autowired
	private SystemService systemService;
	
	private Lock lock = new ReentrantLock();

	@SuppressWarnings(value = {"unchecked", "rawtypes"})
	@Scheduled(cron = "0/30 * * * * ?")
	public void sendOrderService() {
		logger.info("------------task start-----------");
		// If the timer has not stopped, then the next time the timer can not start.
		lock.lock();
		
		try {
			// step 1, Gets the message that is not executed in the database.
			List<MQMessage> messages = mQSerivce.selectWaitHandlerMessage();
			if (messages == null || messages.size() <= 0) {
				return;
			}
			
			// step 2, Handle 10 messages at a time.
			for (MQMessage mqMessage : messages) {
				try {
					Object object = SpringContextHolder.getBean(mqMessage.getServiceHandlerObjectName());
					Class classes = object.getClass();
					Method method = classes.getMethod(mqMessage.getServiceHandlerMethodName(), MQMessage.class);
					if (method == null) {
						mqMessage.setServiceErrorMessage("Method is not defined:" + mqMessage.getServiceHandlerMethodName());
						mqMessage.setServiceStatus(2);
						continue;
					}
					// invoke Method.
					Object resultObject = method.invoke(object, mqMessage);
					if (!(resultObject instanceof Boolean)) {
						mqMessage.setServiceStatus(2);
						mqMessage.setServiceErrorMessage(mqMessage.getServiceHandlerObjectName() + "." + mqMessage.getServiceHandlerMethodName() + "()Return value is not type Boolean");
						continue;
					}
					if (!(Boolean) resultObject) {
						mqMessage.setServiceErrorMessage("invoke Method Error");
						mqMessage.setServiceStatus(2);
						continue;
					}
					mqMessage.setServiceStatus(1);
					continue;
				} catch (Exception e) {
					// TODO: handle exception
					mqMessage.setServiceErrorMessage(e.getMessage());
					mqMessage.setServiceStatus(2);
					continue;
				} finally {
					mqMessage.setServiceIsSend(1);
				}
			}
			
			// step 3, update Message status and serviceIsSend.
			mQSerivce.updateBatchMessage(messages);
			
			// step 4, is next ?
			if (messages.size() >= 10) {
				StaticCacheMemory.isStartTask = true;
				this.sendOrderService();
			}
		} catch (Exception e) {
			e.printStackTrace();
			logger.error(e, "MQ Task Error:" + e.getMessage());
		} finally {
			// reset lock
			lock.unlock();
		}
		logger.info("------------task invoke success-----------");
	}

	/**
	 * 根据订单号拉取物流信息
	 * @throws Exception
	 */
	@Scheduled(cron = "0 0 */1 * * ?")//每小时执行一次
	//@Scheduled(cron = "0 0/2 * * * ?")//每2分钟执行一次
	//@Scheduled(cron = "*/10 * * * * ?")
	public void syncLgisticsInfoByOrderId() throws Exception {
		logger.info("---------------- 根据订单号拉取物流信息开始 ---------------");
		lock.lock();
		List<Order> orderList = null;
		String url = NameUtils.getConfig("rst_efast_base_url");
		String sdId = NameUtils.getConfig("sd_id");
		Map<String, Object> orderData = new HashMap<String, Object>();
		Map<String, String> requestData = new HashMap<String, String>();
		com.fasterxml.jackson.databind.ObjectMapper mapper = new ObjectMapper();
		try {
			logger.info("---------------- selectProbationShopOrderList begin ---------------");
			orderList = probationShopOrderService.selectWateroPFOrderList();
			if(orderList != null && orderList.size() > 0){
				for (Order order:orderList) {
					String salesOrderid = order.getSalesOrderid();
					orderData.put("orderId", salesOrderid);
					orderData.put("sd_id", sdId);
					requestData.put("app_act", "rst.trade.logistics.get");
					String orderDataStr = mapper.writeValueAsString(orderData);
					requestData.put("info", orderDataStr);
					logger.info("app_act==:" + requestData.get("app_act") + "info==:" + requestData.get("info"));
					String result = HttpClient431Util.doPost(requestData, url);
					result = StringEscapeUtils.unescapeJava(result); // unicode 编码
					logger.info("拉取订单号为:" + order.getSalesOrderid() + "的订单，rst返回信息:" + result);
					JSONObject jsonObject = JSONObject.fromObject(result);
					String msg = jsonObject.getString("msg");
					logger.info("---------"+msg+"---------");
					if ("success".equals(msg)) {
						Date date = new Date();
						String salesPostFirm = jsonObject.getString("salesPostFirm");
						String salesPostNum = jsonObject.getString("salesPostNum");
						order.setSalesPostFirm(salesPostFirm);
						order.setSalesPostNum(salesPostNum);
						order.setSalesSendDate(date);
						probationShopOrderService.updateWateroPFOrder(order);
					}
				}
			}
		}catch (Exception e){
			logger.error("拉取订单异常",e.getMessage());
		}finally {lock.unlock();
		}

		logger.info("---------------- 根据订单号拉取物流信息结束 ---------------");
	}


	/**
	 * 订阅消息
	 */
	public void subscribeMessageQueue() {
	}
	
	// 强迫线程可见
	private volatile boolean oldState = true;
	
	/**
	 * 微信定期任务奖励-上个月的定时器
	 */
	@SuppressWarnings("deprecation")
//	@Scheduled(cron = "0/59 * * * * ?")
	public void oldMonthTask() {
		if (!oldState) {
			logger.info("线程正忙...");
			return;
		}
		oldState = false;
		try {
			// 判断上一个月的数据是否奖励完成
			logger.info("上一个月的定时奖励任务奖励...");
			// 当前时间
			Date nowDate = new Date();
			
			// 回到上一个月
			nowDate.setMonth(nowDate.getMonth() - 1);
			// 上一个月的结束时间  类似 ：2016-02-29 23:59:59
			Date oldMonthLastDay = DateTimeUtil.getLastDayOfYear(nowDate);
			// 上一个月的开始时间  类似 ：2016-02-01 00:00:00
			Date oldMonthFirstDay = DateTimeUtil.getFirstDayOfYear(nowDate);
			
			List<WechatTask> tasks = wechatTaskService.getRunTask(oldMonthFirstDay, oldMonthLastDay);
			if (tasks == null || tasks.size() == 0) {
				logger.info("暂时没有上一个月的任务奖励...");
				oldState = true;
				return;
			}
			// 获取在上一个月销售达到X台的数据
			for (WechatTask wechatTask : tasks) {
				List<TaskModel> models = rebackServices.getByDate(wechatTask.getTaskProductType(), oldMonthFirstDay, oldMonthLastDay, wechatTask.getTaskSalesNum());
				if (models == null || models.size() == 0) {
					logger.info("暂时没有人完成上一个月的任务" + wechatTask.getTaskId() + "奖励...");
					continue;
				}
				// 达到的用户是否已经领取奖励
				for (TaskModel taskModel : models) {
					WechatTaskLogs logs = new WechatTaskLogs();
					logs.setLogsTaskId(wechatTask.getTaskId());
					logs.setLogsUserOpenId(taskModel.getOpenID());
					Integer logsID = wechatTaskService.getByTaskIdAndOpenId(logs);
					if (logsID != null) {
						// 如果当前用户已经领取了奖励，那么跳过
						continue;
					}
					// 若用没有领取，那么奖励
					try {
						wechatTaskService.handlerTask(taskModel.getOpenID(), wechatTask.getTaskMoney(), wechatTask.getTaskId(), wechatTask.getTaskProductType());
					} catch (Exception e) {
						logger.error(this, "对:" + taskModel.getOpenID() + ",奖励失败!error:" + e.getMessage());
					}
				}
			}
		} catch (Exception e) {
			logger.error(this, "发生异常,奖励失败!error:" + e.getMessage());
		} finally {
			// 归还锁
			oldState = true;
		}
	}

	// 强迫线程可见
	private volatile boolean nowState = true;
	/**
	 * 微信定期任务奖励-本月的定时器
	 */
//	@Scheduled(cron = "0/59 * * * * ?")
	public void nowMonthTask() {
		if (!nowState) {
			logger.info("线程正忙...");
			return;
		}
		nowState = false;
		
		try {
			// 判断本月是否有奖励数据
			logger.info("本月的定时奖励任务奖励...");
			// 当前时间
			Date nowDate = new Date();
			
			// 上一个月的结束时间  类似 ：2016-02-29 23:59:59
			Date oldMonthLastDay = DateTimeUtil.getLastDayOfYear(nowDate);
			// 上一个月的开始时间  类似 ：2016-02-01 00:00:00
			Date oldMonthFirstDay = DateTimeUtil.getFirstDayOfYear(nowDate);
			
			List<WechatTask> tasks = wechatTaskService.getRunTask(oldMonthFirstDay, oldMonthLastDay);
			if (tasks == null || tasks.size() == 0) {
				logger.info("暂时没有本月的任务奖励...");
				nowState = true;
				return;
			}
			// 获取在上一个月销售达到X台的数据
			for (WechatTask wechatTask : tasks) {
				List<TaskModel> models = rebackServices.getByDate(wechatTask.getTaskProductType(), oldMonthFirstDay, oldMonthLastDay, wechatTask.getTaskSalesNum());
				if (models == null || models.size() == 0) {
					logger.info("暂时没有人完成本月的任务" + wechatTask.getTaskId() + "奖励...");
					continue;
				}
				// 达到的用户是否已经领取奖励
				for (TaskModel taskModel : models) {
					WechatTaskLogs logs = new WechatTaskLogs();
					logs.setLogsTaskId(wechatTask.getTaskId());
					logs.setLogsUserOpenId(taskModel.getOpenID());
					Integer logsID = wechatTaskService.getByTaskIdAndOpenId(logs);
					if (logsID != null) {
						// 如果当前用户已经领取了奖励，那么跳过
						continue;
					}
					// 若用没有领取，那么奖励
					try {
						wechatTaskService.handlerTask(taskModel.getOpenID(), wechatTask.getTaskMoney(), wechatTask.getTaskId(), wechatTask.getTaskProductType());
					} catch (Exception e) {
						logger.error(this, "对:" + taskModel.getOpenID() + ",奖励失败!error:" + e.getMessage());
					}
				}
			}
		} catch (Exception e) {
			// TODO: handle exception
			logger.error(this, "发生异常:" + e.getMessage());
		} finally {
			nowState = true;
		}
	}
	
	public void afterPropertiesSet() throws Exception {
	}
	
	/**
	 * 试用订单短信推送服务（满足XX天奖励XX奖励）
	 */
//	@Scheduled(cron = "0 0 */1 * * ?")//每小时执行一次
	public void probationRewardPush(){
		logger.info("------------执行推送满足奖励条件用户 start-----------"); 
		//查询奖励推送规则
		List<ProbationAwardRelu> reluList = probationShopOrderService.selectProbationAwardRelu(); 
		 SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		if(reluList!=null && !reluList.isEmpty()){
			for(ProbationAwardRelu awardRelu : reluList){
				//查询所有待支付订单
				List<ProbationOrderDto> orderDtoList = probationShopOrderService.selectProbationStartOrder();
				if(orderDtoList!=null && !orderDtoList.isEmpty()){
					for(ProbationOrderDto orderDto : orderDtoList ){
						//计算总试用期限(小时)
						Long totalDate = (orderDto.getProbationEndTime().getTime() - orderDto.getProbationStartTime().getTime()) / 1000 / 60 / 60;
						//计算试用剩余期限(小时)
						Long remainingDate = (orderDto.getProbationEndTime().getTime() - new Date().getTime()) / 1000 / 60 / 60;
						//暂停时间
						/*Long pauseDate = 0l;
						if(orderDto.getProbationPauseStartTime() != null && orderDto.getProbationPauseEndTime() != null){
							pauseDate =	(orderDto.getProbationPauseEndTime().getTime() - orderDto.getProbationPauseStartTime().getTime()) / 1000 / 60 / 60;
						}*/
						//实际试用时间 (总试用时间 - 试用剩余期限 - 暂停时间)
 						Long actualProbationDate = totalDate - remainingDate - orderDto.getProbationPauseTotalLength() / 1000 / 60 / 60;
						//奖励奖品天数转换为小时
						Long awardDate = Long.valueOf(awardRelu.getAwardNum()) * 24;
						//奖励条件时间 - 实际试用时间  （大于23小时或者小于26小时则推送）
						if( awardDate -actualProbationDate > 23 && awardDate -actualProbationDate < 26 ){
							ProbationNewLogs probationNewLogs =	probationShopOrderService.selectProbationNewLogsByOrderId(orderDto.getProbationOrderid());
							//添加日志信息
							ProbationNewLogs newLogs = new ProbationNewLogs();
							newLogs.setLogsPorderid(orderDto.getProbationOrderid());
							newLogs.setLogsTime(awardRelu.getAwardNum());
							newLogs.setLogsCreateTime(new Date());
							if(probationNewLogs == null){
								//推送操作
								sendMessageUtil.probationPush(awardRelu.getAwardInfo(), 
										orderDto.getProductName(), 
										orderDto.getProbationOrderid(), 
										formatter.format(orderDto.getProbationStartTime()), 
										formatter.format(
												new Date(
														orderDto.getProbationStartTime().getTime() + 
														(awardRelu.getAwardNum() * 24 + orderDto.getProbationPauseTotalLength()) * 60 * 60 * 1000)), 
										awardRelu.getAwardInfoTwo(), 
										orderDto.getProbationOpenid(), 
										ResultInfo.TRY_ORDER_DETAIL+"?orderId="+orderDto.getProbationOrderid());
								//添加日志
								probationShopOrderService.insertProbationLogs(newLogs);
								//发送短信给收货人
								if(StringUtils.isNotEmpty(orderDto.getProbationReceiveTel())){
									codeService.sendTextToUser(orderDto.getProbationReceiveTel(), 6);
								}
							}else if(probationNewLogs.getLogsTime() < awardRelu.getAwardNum()){
								//执行推送
								sendMessageUtil.probationPush(
										awardRelu.getAwardInfo(), 
										orderDto.getProductName(), 
										orderDto.getProbationOrderid(), 
										formatter.format(orderDto.getProbationStartTime()), 
										formatter.format(
												new Date(orderDto.getProbationStartTime().getTime() 
														+ (awardRelu.getAwardNum() * 24 + orderDto.getProbationPauseTotalLength()) * 60 * 60 * 1000)),
										awardRelu.getAwardInfoTwo(), 
										orderDto.getProbationOpenid(), 
										ResultInfo.TRY_ORDER_DETAIL+"?orderId="+orderDto.getProbationOrderid());
								//修改日志
								probationShopOrderService.updatePauseLogsDate(newLogs);
								//发送短信给收货人
								if(StringUtils.isNotEmpty(orderDto.getProbationReceiveTel())){
									codeService.sendTextToUser(orderDto.getProbationReceiveTel(), awardRelu.getAwardNum().intValue());
								}
							}
						}
					}
				}
			}
		}
	}

	@SuppressWarnings(value = {"unchecked", "rawtypes"})
	@Scheduled(cron = "0 0/1 * * * ?")
	public void selectEfastOrder() {
		try{
			//获取cookie值
			ShopSystemRule rule =  systemService.selectOneShopRuleById(244);
			String phone = "13590159563";
			String url = NameUtils.getConfig("efast_order_url") + phone;
			Map<String,String> params = new HashMap<String,String>();
			String cookie = rule.getRuleDesc();
			//查询百胜订单信息
			String str = HttpClient431Util.doGet(params,url,null,null,cookie);
		} catch(Exception e) {
			System.out.println("------------请求百胜接口失败-------------");
		}

	}
}