用Elastic-Job可解决分布式重复执行问题

如果业务工程采用集群化的部署，可能会多次重复执行定时任务而导致系统的业务逻辑错误，并产生系统故障。

job.properties

simple.id=recommendJobsimple.class=com.tf56.mk.job.RecommendJob#每天23点执行simple.cron=0 0 23 * * ? *##0 0 23 * * ? *##0 0/1 * * * ?## 分片，会根据分片数配置多线程处理。在批量数据处理时，分片处理会提高效率。simple.shardingTotalCount=1## 分片参数配置，使得分片更易于理解，某些时候可易于作业处理。simple.shardingItemParameters=0=Beijing## 是否监控作业运行状态。在任务执行时间短、任务间隔时间短等情况下，不建议开启，多少会影响效率。simple.monitorExecution=true## 失效转移。在多服务器时比较有用。simple.failover=true## 作业描述simple.description=MK简单作业## 作业是否禁止启动。可用于部署作业时，先禁止启动，部署结束后统一启动。simple.disabled=false## 本地配置是否可覆盖注册中心配置。如果可覆盖，每次启动作业都以本地配置为准simple.overwrite=true## 作业监控端口。建议开启，方便dump作业信息，排查问题，特别是分布式作业情况下。simple.monitorPort=9888## 作业前置、后置处理监听器。有点像是拦截器的概念。listener.simple=com.tf56.mk.job.listener.SimpleListenersimple1.id=projectJobsimple1.class=com.tf56.mk.job.ProjectJob##0 0 1 1/1 * ?  每日1点执行##0 0/5 * * * ?  每5分钟执行simple1.cron=0 0/5 * * * ?

job.xml

reg.properites

serverLists=mt-zookeeper-vip:2181namespace=esjob-monkeyKingServicebaseSleepTimeMilliseconds=1000maxSleepTimeMilliseconds=3000maxRetries=3

package com.tf56.mk.job;import com.dangdang.ddframe.job.api.ShardingContext;import com.dangdang.ddframe.job.api.simple.SimpleJob;import com.tf56.mk.common.util.PropertiesUtil;import com.tf56.mk.dao.RequestRecommendDao;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.HashMap;import java.util.Map;@Componentpublic class RecommendJob implements SimpleJob {    private final Logger logger = LoggerFactory.getLogger(this.getClass());    @Autowired    RequestRecommendDao requestRecommendDao;    /**     *      * @param arg0     */    @Override    public void execute(ShardingContext arg0) {        logger.info("ES Job Doing--doRecommendJob");        Map map = new HashMap();        map.put("mkExpertRecommendJobTime",Long.parseLong(PropertiesUtil.getPropertieValue("mkExpertRecommendJobTime")));        map.put("mkExpertRecommendNum",Long.parseLong(PropertiesUtil.getPropertieValue("mkExpertRecommendNum")));        requestRecommendDao.doRecommendJob(map);    }}

package com.tf56.mk.job.listener;import com.dangdang.ddframe.job.executor.ShardingContexts;import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;import org.slf4j.Logger;import org.slf4j.LoggerFactory;public class SimpleListener implements ElasticJobListener {    private static Logger logger = LoggerFactory.getLogger(SimpleListener.class);    public void beforeJobExecuted(final ShardingContexts shardingContexts) {        logger.info("beforeJobExecuted:" + shardingContexts.getJobName());    }    public void afterJobExecuted(final ShardingContexts shardingContexts) {        logger.info("afterJobExecuted:" + shardingContexts.getJobName());    }}

pom.xml

        
     
      UTF-8
         
     
      1.7
         
     
      2.0.1
     
    

    
        
     
      com.dangdang
         
     
      elastic-job-lite-core
         
     
      ${elastic-job.version}
         
             
                  
       
        org.eclipse.jetty.orbit
                   
       
        javax.annotation
               
          
     
    
        
     
      com.dangdang
         
     
      elastic-job-lite-spring
         
     
      ${elastic-job.version}
     
    

JobConfig.java 引入xml配置

@Configuration@ImportResource(locations = {"classpath:spring/job.xml"})public class JobConfig {}