spring /quartz集成

pom.xml

<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.0</version>
</dependency>

spring-quartz.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xsi:schemaLocation="http://www.springframework.org/schema/beans    
                        http://www.springframework.org/schema/beans/spring-beans-4.0.xsd    
                        http://www.springframework.org/schema/context    
                        http://www.springframework.org/schema/context/spring-context-4.0.xsd    
                        http://www.springframework.org/schema/mvc    
    
                        http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd">
    <bean name="jobDetail"
        class="org.springframework.scheduling.quartz.JobDetailFactoryBean">
        <property name="jobClass"
            value="com.pptv.pms.admin.service.impl.MessageCountJobFactory" />
        <property name="durability" value="true" />
    </bean>

    <bean id="cronTrigger"
        class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
        <property name="jobDetail" ref="jobDetail" />
        <!-- 每30分钟统计一次 -->
        <property name="cronExpression" value="0 0/30 * * * ?" />
    </bean>

    <bean id="jobFactory" class="com.pptv.pms.admin.common.util.JobFactory"></bean>
    <bean id="schedulerFactoryBean"
        class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
        <property name="jobDetails">
            <list>
                <ref bean="jobDetail" />
            </list>
        </property>
        <property name="triggers">
            <list>
                <ref bean="cronTrigger" />
            </list>
        </property>
        <property name="jobFactory" ref="jobFactory"></property>
    </bean>

</beans>

JobFactory.class

package com.pptv.pms.admin.common.util;

import org.quartz.spi.TriggerFiredBundle;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.scheduling.quartz.AdaptableJobFactory;

/**
 * Created by st
 */
public class JobFactory extends AdaptableJobFactory {

    // 这个对象Spring会帮我们自动注入进来,也属于Spring技术范畴.
    @Autowired
    private AutowireCapableBeanFactory capableBeanFactory;

    protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception {
        // 调用父类的方法
        Object jobInstance = super.createJobInstance(bundle);
        // 进行注入,这属于Spring的技术,不清楚的可以查看Spring的API.
        capableBeanFactory.autowireBean(jobInstance);
        return jobInstance;
    }
}

MessageCountJobFactory.class

package com.pptv.pms.admin.service.impl;

import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.quartz.QuartzJobBean;

import com.pptv.pms.admin.service.IMessageCountService;
import com.pptv.pms.admin.service.IMessageJobService;
import com.sun.swing.internal.plaf.metal.resources.metal;

/**
 * 定时统计消息数据
 * @author st
 *
 */
public class MessageCountJobFactory extends QuartzJobBean {
    
    @Autowired
    private IMessageCountService messageCountService;

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        messageCountService.timerTaskUpdateMessageCount();//定时更新统计数据
    }

}

手动添加任务：

package com.pptv.pms.admin.service.impl;

import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.Date;
import java.util.List;

import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.pptv.pms.admin.common.consts.Final;
import com.pptv.pms.admin.common.listener.load.LoadData;
import com.pptv.pms.admin.common.util.DateUtil;
import com.pptv.pms.admin.common.util.HttpUtil;
import com.pptv.pms.admin.common.util.PushResponse;
import com.pptv.pms.admin.common.util.ReturnEnum;
import com.pptv.pms.admin.common.util.SendStausEnum;
import com.pptv.pms.admin.common.util.StringUtil;
import com.pptv.pms.admin.entity.Dictionary;
import com.pptv.pms.admin.entity.Message;
import com.pptv.pms.admin.service.IMessageCountService;
import com.pptv.pms.admin.service.IMessageJobService;
import com.pptv.pms.admin.service.IMessageService;

@Service(value=MessageJobServiceImpl.serviceName)
public class MessageJobServiceImpl implements IMessageJobService {
    private static Logger log = LoggerFactory.getLogger(MessageJobServiceImpl.class);
    public final static String serviceName="messageJobService";
    @Autowired
    private Scheduler scheduler;
    @Autowired
    private IMessageCountService messageCountService;
    @Autowired
    private IMessageService messageService;
    @Value(value = "${cancel.url}")
    private String cancelUrl;
    @Value(value = "${push2tag.url}")
    private String push2tagUrl;
    @Value(value = "${push2all.url}")
    private String push2AllUrl;
    @Override
    public String addJob(Message message) {
        JSONObject obj = new JSONObject();
        String code = null;
        String msg = null;
        
        if (message.getStatus()==Final.MESSAGE_AUDIT_STATUS_NO_PASS) {//该消息未通过审核不能发送
            code=ReturnEnum.ERROR.getText();
            msg="该消息未通过审核不能发送";// 任务已经过时
            obj.put(Final.AJAX_KEY_CODE, code);
            obj.put(Final.AJAX_KEY_MSG, msg);
            return obj.toString();
        }
        
        if (DateUtil.compare_date(message.getEndTime(), new Date()) == -1) {
            code=ReturnEnum.ERROR.getText();
            msg="该消息已经过期";// 任务已经过时
            obj.put(Final.AJAX_KEY_CODE, code);
            obj.put(Final.AJAX_KEY_MSG, msg);
            return obj.toString();
        }
        if (message.getSendStatus() != null&&message.getTaskid()!=null) {// 已经发送过了
            code=ReturnEnum.ERROR.getText();
            msg="该消息已经发送";// 任务已经过时
            obj.put(Final.AJAX_KEY_CODE, code);
            obj.put(Final.AJAX_KEY_MSG, msg);
            return obj.toString();
        }
        JobKey key = new JobKey(message.getId().toString(), message.getId().toString());
        JobDetail jobDetail = JobBuilder.newJob(MessageJobFactory.class).withIdentity(key).build();
        jobDetail.getJobDataMap().put("message", message);
        SimpleScheduleBuilder builder = SimpleScheduleBuilder.simpleSchedule();
        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity(message.getId().toString(), message.getId().toString()).withSchedule(builder)
                .startAt(message.getStartTime()).endAt(message.getEndTime()).build();
        try {
            scheduler.scheduleJob(jobDetail, trigger);
            messageService.updateMessageSendStatus(message.getId(), SendStausEnum.SEND_PROCESS.name());//更新状态为正在进行中
            messageCountService.updateMessageCountByMessage(message);//及时更新统计数据
            code=ReturnEnum.OK.getText();
            msg="任务开启成功";// 任务开启成功
            obj.put(Final.AJAX_KEY_CODE, code);
            obj.put(Final.AJAX_KEY_MSG, msg);
            return obj.toString();
        } catch (org.quartz.ObjectAlreadyExistsException e2) {
            e2.printStackTrace();
            code=ReturnEnum.ERROR.getText();
            msg="该消息任务已经开启,无需再发送";//该消息任务已经开启,无需再发送
            obj.put(Final.AJAX_KEY_CODE, code);
            obj.put(Final.AJAX_KEY_MSG, msg);
            return obj.toString();
        } catch (SchedulerException e) {
            code=ReturnEnum.ERROR.getText();
            msg=Final.AJAX_KEY_MSG_SYS_ERROR;// 系统异常
            obj.put(Final.AJAX_KEY_CODE, code);
            obj.put(Final.AJAX_KEY_MSG, msg);
            return obj.toString();
        }

    }

    @Override
    public String cancelMessage(Integer id) {

        JSONObject obj = new JSONObject();
        String code = null;
        String msg = null;
        Message message = messageService.getById(id);
        delJob(message);// 首先删除任务
        if (StringUtil.isBlank(message.getTaskid())) {
            messageService.updateMessageSendStatus(message.getId(), SendStausEnum.CANCEL_SUCCESS.name());
            code = ReturnEnum.OK.getText();
            msg = "取消成功";
            obj.put(Final.AJAX_KEY_CODE, code);
            obj.put(Final.AJAX_KEY_MSG, msg);
            return obj.toString();
        }
        try {
            log.info("----------开始执行撤销任务--------------");
            Dictionary appIdDictionary = LoadData.getDicDataByKey(Final.DB_PUSH_APPID);
            Dictionary appsecretDictionary = LoadData.getDicDataByKey(Final.DB_PUSH_APPSECRET);
            String reponseJson = HttpUtil.doPostSSL(cancelUrl, "appid=" + appIdDictionary.getItemValue() + "&appsecret="
                    + appsecretDictionary.getItemValue() + "&taskid=" + message.getTaskid());
            PushResponse response = JSONObject.toJavaObject(JSON.parseObject(reponseJson), PushResponse.class);
            if (response.getStatusCode() == 1000) {// 撤销成功
                messageService.updateMessageSendStatus(message.getId(), SendStausEnum.CANCEL_SUCCESS.name());
                messageCountService.updateMessageCountByMessage(message);//及时更新统计数据
                code = ReturnEnum.OK.getText();
                msg = "撤销成功";
            }else {// 撤销失败
                messageService.updateMessageSendStatus(message.getId(), SendStausEnum.CANCEL_FAIL.name());
                code = ReturnEnum.ERROR.getText();
                msg = response.getStatusDesc();
            }
        } catch (Exception e) {//// 撤销失败
            messageService.updateMessageSendStatus(message.getId(), SendStausEnum.CANCEL_FAIL.name());
            code = ReturnEnum.ERROR.getText();
            msg = "撤销失败";
        }
        obj.put(Final.AJAX_KEY_CODE, code);
        obj.put(Final.AJAX_KEY_MSG, msg);
        return obj.toString();
    }

    /**
     * 重新发送一次任务
     */
    @Override
    public String reSendMessage(Integer id) {
        JSONObject obj = new JSONObject();
        String code = null;
        String msg = null;
        Message message = messageService.getById(id);
        if (message.getSendStatus() != null && message.getSendStatus().equals(SendStausEnum.SEND_SUCCESS.name())) {// 已经发送
            code = ReturnEnum.ERROR.getText();
            msg = "该消息已经发送，无需再发送";
            obj.put(Final.AJAX_KEY_CODE, code);
            obj.put(Final.AJAX_KEY_MSG, msg);
            return obj.toString();
        } else {
            return reSendMessage(message);
        }

    }

    /**
     * 执行任务
     */
    @Override
    public String reSendMessage(Message message) {
        JSONObject obj = new JSONObject();
        String code = null;
        String msg = null;
        try {
            // int a =1/0;
            log.info("----------开始执行任务--------------");
            Dictionary appIdDictionary = LoadData.getDicDataByKey(Final.DB_PUSH_APPID);
            Dictionary appsecretDictionary = LoadData.getDicDataByKey(Final.DB_PUSH_APPSECRET);
            String url = push2tagUrl;
            if (message.getAppName()
                    .equals(LoadData.getDicDataByKey(Final.DB_MESSAGE_DEVAPP_TYPE_ALL).getItemNamecn())) {
                url = push2AllUrl;
            }
            
            
            String msgCxt =message.getMessage();
            try {
                msgCxt=URLEncoder.encode(msgCxt.toString(), "UTF-8");
                System.out.println(msgCxt);
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                log.error(e.getMessage());
            }
            
            String reponseJson = HttpUtil.doPostSSL(url,
                    "appid=" + appIdDictionary.getItemValue() + "&appsecret=" + appsecretDictionary.getItemValue()
                            + "&msg=" + msgCxt + "&tag=" + message.getTagName() + "&timeout="
                            + message.getEndTime().getTime());
            PushResponse response = JSONObject.toJavaObject(JSON.parseObject(reponseJson), PushResponse.class);
            if (response.getStatusCode() == 1000) {// 成功
                messageService.updateMessageSendStatus(message.getId(), SendStausEnum.SEND_SUCCESS.name());
                messageService.updateMessageTaskId(message.getId(), response.getResult());
                messageCountService.updateMessageCountByMessage(message);//及时更新统计数据
                code = ReturnEnum.OK.getText();
                msg = "发送成功";
            }else {//失败
                messageService.updateMessageSendStatus(message.getId(), SendStausEnum.SEND_FAIL.name());
                code = ReturnEnum.ERROR.getText();
                msg = response.getStatusDesc();
            }

        } catch (Exception e) {
            log.error("error:" + e.getMessage());
            messageService.updateMessageSendStatus(message.getId(), SendStausEnum.SEND_FAIL.name());
            code = ReturnEnum.ERROR.getText();
            msg = Final.AJAX_KEY_MSG_SYS_ERROR;
            e.printStackTrace();
        }

        obj.put(Final.AJAX_KEY_CODE, code);
        obj.put(Final.AJAX_KEY_MSG, msg);
        return obj.toString();

    }

    @Override
    public String startTaskMessage(Integer id) {
        String json =addJob(messageService.getById(id));
        return json;
    }
    
    @Override
    public void delJob(Message message) {
        JobKey key = new JobKey(message.getId().toString(), message.getId().toString());
        try {
            scheduler.deleteJob(key);
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void initMessageJob() {
        List<Message> messages = messageService.getPassAndNoEndAndSendProcessMessage();
        for (Message message : messages) {
            addJob(message);
        }

    }
}