spring+redis+mysql定時重發

產品需求:發布信息時候，用戶可以選擇重新發布時間，重新發布間隔時間以及次數，如重發6次時間間隔為2分鍾。

實現設計:

　　　　可以通過定時查詢數據庫的發布時間，以及信息中重發次數時間，update數據(或者直接寫一個厲害的update語句也可以)，但是定時的查詢整個數據庫表，而且中間還會涉及到運算，無法命中索引，會使數據庫壓力較大，咨詢好多朋友，大多數建議降低數據查詢次數，如半個小時update一次，但是這樣時間誤差會很大，產品無法接受。

　　　　在通過仔細調研以之后，計算使用spring定時任務、redis中list結構、update數據實現。當用戶發布信息的時候，依據次數生成多個序列的key，並將信息id存儲到redis中(如果多個信息，在同一個序列中，直接在list尾部添加數據)，spring通過定時任務，每隔10秒依據生成的序列，讀取redis中list數據，將list中數據，轉化為id1，id2，id3，通過數據庫in關鍵字，實現更新數據。

關鍵代碼:

　　　　spring定時任務配置

    <bean id="republishCargoJobInvake" class="org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean">
        <property name="targetObject" ref="republishCargoJob" />
        <property name="targetMethod" value="doRepublish" />
        <property name="concurrent" value="false" />
    </bean>
        <bean id="triggerRepublishCargo" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
        <property name="jobDetail" ref="republishCargoJobInvake" />    
        <property name="cronExpression" value="0/10 * * * * ?" />

    </bean>
         <bean class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
        <property name="triggers">
            <list>
                <ref bean="triggerRepublishCargo" />
            </list>
        </property>   
    </bean>                

public class RepublishCargoJob {
    private Log log = LogFactory.getLog(RepublishCargoJob.class);
    /**
     * 貨源重發定時任務
     */
    public void doRepublish(){
        //獲取WebApplicationContext
        WebApplicationContext wac = ContextLoader.getCurrentWebApplicationContext();
        //獲取redis讀取序列工具
        TimedPublishTask timedPublishTask=(TimedPublishTask) wac.getBean("timedPublishTask");
        //獲取本次需要更新的id list
        List<String>list=timedPublishTask.getBatch();
        if(list!=null&&list.size()>0){
            //update數據庫
            ResourceCargoInfoDAO resourceCargoInfoDAO=(ResourceCargoInfoDAO) wac.getBean("resourceCargoInfoDAO");
            ResourceCargoInfo rci=new ResourceCargoInfo();
            rci.setRciPublishDatetime(DateUtils.getDateTime());
            ResourceCargoInfoExample resourceCargoInfoExample=new ResourceCargoInfoExample();
            resourceCargoInfoExample.createCriteria().andRciIdIn(list).andRciAvalibleStatusEqualTo(1);
        try {
            resourceCargoInfoDAO.updateByExampleSelective(rci, resourceCargoInfoExample);
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            log.error("=======重發貨源失敗======");
        }
        }
    }

package com.ada.wuliu.common.redis.timertask;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.jedis.JedisConnection;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.InitBinder;

import com.ada.wuliu.common.dictionaries.utils.DateUtils;
import com.ada.wuliu.common.dictionaries.utils.StateCode;
import com.ada.wuliu.common.dto.result.SimpleResult;
import com.ada.wuliu.common.redis.dao.RedisList;
import com.ada.wuliu.common.redis.dao.RedisString;
import com.ada.wuliu.common.redis.dto.TimerPublish;
import com.ada.wuliu.common.redis.util.SessionImpl_Temp_01;
import com.ada.wuliu.common.redis.util.StringUtil;
@Component("timedPublishTask")
public class TimedPublishTask {
    //定時任務執行周期
    public Integer taskFrequency=10000;
    //默認任務開始\(^o^)/
    private String defaultIncr="1000";

    private static String TIMER_PUBLISH_KEY_TODO=null;

    @Autowired
    private RedisList redisList;
    @Autowired
    private RedisString redisString;
    private final static Log log = LogFactory.getLog(TimedPublishTask.class);
    @Resource(name="connectionFactory")
    private JedisConnectionFactory connectionFactory;
    public TimedPublishTask(){
        initKeyHeader();
    }
    private void initKeyHeader(){
        if(TIMER_PUBLISH_KEY_TODO==null){
            TIMER_PUBLISH_KEY_TODO="TIMER_PUBLISH_KEY_TODO_CYC_"+StringUtil.getLocalOsName();
        }
    }
    /**
     * 
     * @param timerPublish 需要重新發布的對象
     */
    public  SimpleResult addBatch(TimerPublish timerPublish){
        JedisConnection connection = connectionFactory.getConnection();
        SimpleResult cr = null;
        try {
            //獲取最新的批次號
            String batchNo= redisString.get(TIMER_PUBLISH_KEY_TODO, connection);
            if(batchNo==null){
                batchNo=redisString.incr(TIMER_PUBLISH_KEY_TODO, connection).toString();
            }
            //刷新頻率
            Integer frequency=timerPublish.getFrequency()/10;
            //刷新次數
            Integer number=timerPublish.getNumber();
            //對象key
            Long key=timerPublish.getKey();
            List<Long>list=new ArrayList<Long>();
            for(int i=1;i<number+1;i++){
                //生成添加的批次號
                long addBatchNo=i*frequency/taskFrequency+Long.parseLong(batchNo);
                addBatchNo=addBatchNo-1;
                String batchKey=TIMER_PUBLISH_KEY_TODO+addBatchNo;
                //list 第一個為更新時間 如果更新時間<當前時間10秒則updatelist索引key對應數據
                log.info("新添加批次號:======"+batchKey+"====="+key);
                List<String> ll=redisList.get(batchKey, 1, 2, connection);
            
                if(ll==null||ll.size()==0){
                
                    redisList.lPush(batchKey, (System.currentTimeMillis()+i*frequency)+"", connection);    
                }
                redisList.rPush(batchKey, key+"", connection);    
            
            }
            return new SimpleResult("成功", StateCode.SUCCESS, StateCode.SUCCESS, true);
        } catch (Exception e) {
            log.error("-----TimedPublishTask::addBatch Throwable Error! -----", e);
            return new SimpleResult("服務器錯誤", StateCode.SERVER_ERROR, null, false);
        } catch (Throwable th) {
            log.error("-----TimedPublishTask::addBatch Throwable Error! -----", th);
            return new SimpleResult("服務器錯誤", StateCode.SERVER_ERROR, null, false);
        } finally {
            connection.close();
        }
    }
    
    /**
     * 
     * @param batchNo 獲取本批次需要更新的對象的key
     * @return
     */
    public  List<String> getBatch(){

        JedisConnection connection = connectionFactory.getConnection();
        SimpleResult cr = null;
        try {
            //獲取當前需要更新的批次號
            String batchNo= redisString.get(TIMER_PUBLISH_KEY_TODO, connection);
            //如果沒有批次則初始化批次
            if(batchNo==null){
                batchNo=redisString.incr(TIMER_PUBLISH_KEY_TODO, connection).toString();
            }
            List<String>list=redisList.getAll(TIMER_PUBLISH_KEY_TODO+batchNo, connection);
            log.info("當前任務"+TIMER_PUBLISH_KEY_TODO+batchNo);
            //出現掉任務處理
            if(list!=null&&list.size()>0){
                Long addTime=Long.parseLong(list.get(0));
                Long nowTime=System.currentTimeMillis();
                if(nowTime-addTime>2*taskFrequency){
                    log.info("發現隊列冗余數據，開始嘗試執行=========");
                    //獲取偏移量
                    long offset=(nowTime-addTime)/taskFrequency+1;
                    log.info("偏移量======"+offset);
                    Integer batchNoRe=Integer.parseInt(batchNo);
                    int i=1;
                    if(offset<50){ i=1;}
                    else{i=(int) (offset-50L);}
                    for(;i<offset;i++){
                        batchNoRe++;
                        log.info("添加任務"+TIMER_PUBLISH_KEY_TODO+batchNoRe);
                        List<String>listre=redisList.getAll(TIMER_PUBLISH_KEY_TODO+batchNoRe, connection);
                        if(listre.size()>0) listre.remove(0);
                        redisString.del(TIMER_PUBLISH_KEY_TODO+batchNoRe, connection);
                        redisString.incr(TIMER_PUBLISH_KEY_TODO, connection);
                        //redisString.incr(TIMER_PUBLISH_KEY_TODO, connection);
                        list.addAll(listre);
                    }
                }
            }
            if(list.size()>0) list.remove(0);
            log.info("執行批次==========="+TIMER_PUBLISH_KEY_TODO+batchNo+"======VALUES====="+getListStr(list));
            redisString.incr(TIMER_PUBLISH_KEY_TODO, connection);
            redisString.del(TIMER_PUBLISH_KEY_TODO+batchNo, connection);
            return list;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("-----TimedPublishTask::getBatch Throwable Error! -----", e);
            return null;
        } catch (Throwable th) {
            log.error("-----TimedPublishTask::getBatch Throwable Error! -----", th);
            return null;
        } finally {
            connection.close();
        }
    }
    public String getListStr(List<String>list){
        if(list==null)return null;
        StringBuffer sb=new StringBuffer();
        for(String s:list){
            sb.append(s+",");
        }
        return sb.toString();
    }
    
}

  初始化批次15秒執行一次，需要執行10次

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL1=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL3=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL4=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL6=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL7=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL9=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL10=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL12=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL13=====114102091

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL15=====114102091

初始化批次15秒執行一次，需要執行10次

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL1=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL3=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL4=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL6=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL7=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL9=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL10=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL12=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL13=====114102095

新添加批次號:======TIMER_PUBLISH_KEY_TODO_KTSDKLHL15=====114102095

當前任務TIMER_PUBLISH_KEY_TODO_KTSDKLHLSK1

發現隊列冗余數據，開始嘗試執行=========

偏移量======5

添加任務TIMER_PUBLISH_KEY_TODO_KTSDKLHLSK2

添加任務TIMER_PUBLISH_KEY_TODO_KTSDKLHLSK3

添加任務TIMER_PUBLISH_KEY_TODO_KTSDKLHLSK4

添加任務TIMER_PUBLISH_KEY_TODO_KTSDKLHLSK5

執行批次===========TIMER_PUBLISH_KEY_TODO_KTSDKLHLSK6======VALUES=====114102091,114102095,