【转】java调度器(重试机制)实现

【转】http://blog.csdn.net/prime7/article/details/49837517

调度器分为正常调度，异常调度，异常调度根据不同的队列进行时间间隔的区分，采用ScheduledExecutorService进行时间间隔调度，调度时根据当前队列中addData进队列里的数据分配线程进行处理。先看正常调度，这个类在构造函数时就会被调度，可以添加set方法，配置好单独再进行scheduler：

 

package test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 异步任务定时执行管理
 */
public abstract class AbstractScheduleTask<T> implements ScheduleTask<T> {

    /** 任务队列 */
    protected final BlockingQueue<T> dataQueue;
    /** 线程池 */
    protected final ExecutorService executorService;
    /** 调度任务 */
    protected final ScheduledExecutorService scheduledExecutorService;
    /** 定时任务数 */
    private int scheduleSize = 1;
    /** 定时任务开始执行时间 */
    private int scheduleInitialDelay = 0;
    /** 定时任务间隔时间,正常单条数据的插入时间<3ms,队列的长度为1000,1000m秒的时间足够,避免队列的数据堆积 */
    private int scheduleDelay = 1000;
    /** 线程池大小 */
    private int threadPoolSize = 8;
    /** 队列大小 */
    private int queueSize = 2000;
    /** 线程批处理大小; */
    private int dataSize = 100;

    /** 默认构造方法，加载定时任务 */
    public AbstractScheduleTask() {
        dataQueue = new LinkedBlockingQueue<T>(queueSize);
        executorService = Executors.newFixedThreadPool(threadPoolSize);
        scheduledExecutorService = Executors.newScheduledThreadPool(scheduleSize);
        schedule();
    }

    /**
     * 具体业务数据处理
     *
     * @param data
     * @return
     */
    protected abstract Integer doData(final List<T> data);

    /**
     * 添加数据到队列
     */
    @Override
    public void addData(T parameterObject) {
        if (parameterObject != null) {
            if (dataQueue.size() >= this.getQueueSize()) {
                // 消费队列数据过大
            }
            try {
                dataQueue.put(parameterObject);
            } catch (InterruptedException e) {
                // 添加队列数据异常
            }
        }
    }

    /**
     * 设置定时任务 设定执行线程计划,初始10s延迟,每次任务完成后延迟10s再执行一次任务
     */
    private void schedule() {
        for (int i = 0; i < scheduleSize; i++) {
            scheduledExecutorService.scheduleWithFixedDelay(new ScheduleHandler(), scheduleInitialDelay, scheduleDelay,
                    TimeUnit.MILLISECONDS);
        }

    }

    /**
     * 创建任务
     *
     * @param data
     * @return
     */
    private Callable<Integer> getTask(final List<T> data) {
        Callable<Integer> task = new Callable<Integer>() {

            @Override
            public Integer call() throws Exception {
                if (data == null) {
                    return 0;
                }
                try {
                    int result = doData(data);
                    // 任务执行完成
                    return result;
                } catch (Throwable e) {
                    // 执行任务出现异常
                }
                return 0;
            }
        };

        return task;
    }

    /**
     * 定时任务
     */
    private final class ScheduleHandler implements Runnable {
        /**
         * 定时任务实现
         *
         * @see java.lang.Runnable#run()
         */
        public void run() {
            List<T> dataList = new ArrayList<T>(dataSize);
            while (!dataQueue.isEmpty()) {
                if (dataList.size() == dataSize) {
                    break;
                }
                try {
                    T data = dataQueue.take();
                    dataList.add(data);
                } catch (InterruptedException e) {
                    // 出队列执行异常
                }
            }
            Callable<Integer> task = getTask(dataList);
            if (task == null) {
                return;
            }
            Future<Integer> future = executorService.submit(task);
            try {
                int result = future.get();
                // 任务执行结果: result
            } catch (InterruptedException e) {
                // 任务执行异常
            } catch (ExecutionException e) {
                // 任务执行异常
            }
        }

    }

    /**
     * Getter method for property <tt>scheduleSize</tt>.
     *
     * @return property value of scheduleSize
     */
    public int getScheduleSize() {
        return scheduleSize;
    }

    /**
     * Getter method for property <tt>scheduleInitialDelay</tt>.
     *
     * @return property value of scheduleInitialDelay
     */
    public int getScheduleInitialDelay() {
        return scheduleInitialDelay;
    }

    /**
     * Getter method for property <tt>scheduleDelay</tt>.
     *
     * @return property value of scheduleDelay
     */
    public int getScheduleDelay() {
        return scheduleDelay;
    }

    /**
     * Getter method for property <tt>threadPoolSize</tt>.
     *
     * @return property value of threadPoolSize
     */
    public int getThreadPoolSize() {
        return threadPoolSize;
    }

    /**
     * Getter method for property <tt>queueSize</tt>.
     *
     * @return property value of queueSize
     */
    public int getQueueSize() {
        return queueSize;
    }

    /**
     * Getter method for property <tt>dataSize</tt>.
     *
     * @return property value of dataSize
     */
    public int getDataSize() {
        return dataSize;
    }
}

实现的添加数据接口的定义：

 

 

package test;

/**
 * 任务定时执行管理器
 */
public interface ScheduleTask<T> {

    /**
     * 添加数据
     *
     * @param parameterObject
     */
    public void addData(final T parameterObject);

}

再来看一下 实现带有重试机制的调度器，不同的重试机制使用不同的调度队列

 

 

package test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 重试调度器
 *
 * @author prime7
 */
public abstract class AbstractRetryScheduleTask<T> extends AbstractScheduleTask<T> implements RetryScheduleTask<T> {

    /** 重试调度模式，供外部添加调用 */
    public static final int RETRY_MODE_30S = 1;
    public static final int RETRY_MODE_5MIN = 2;
    public static final int RETRY_MODE_30MIN = 3;

    /** 重试任务队列 */
    protected final BlockingQueue<T> retryDataQueue30S;
    protected final BlockingQueue<T> retryDataQueue5Min;
    protected final BlockingQueue<T> retryDataQueue30Min;

    /** 重试调度任务 */
    protected final ScheduledExecutorService retryScheduledExecutorService;
    /** 重试定时任务数 */
    private int retryScheduleSize = 3;
    /** 重试定时任务开始执行时间 */
    private int retryScheduleInitialDelay = 0;
    /** 重试任务间隔时间 */
    private int retryScheduleDelay30S = 1000; // 30 * 1000;
    private int retryScheduleDelay5Min = 5000; // 5 * 60 * 1000;
    private int retryScheduleDelay30Min = 10000; // 30 * 60 * 1000;
    /** 队列大小 */
    private int retryQueueSize = 2000;
    /** 线程批处理大小; */
    private int retryDataSize = 100;

    public AbstractRetryScheduleTask() {
        super();
        // LinkedBlockingQueue is Thread safety.
        retryDataQueue30S = new LinkedBlockingQueue<T>(retryQueueSize);
        retryDataQueue5Min = new LinkedBlockingQueue<T>(retryQueueSize);
        retryDataQueue30Min = new LinkedBlockingQueue<T>(retryQueueSize);
        retryScheduledExecutorService = Executors.newScheduledThreadPool(retryScheduleSize);
        schedule();
    }

    private void schedule() {
        // init 30s retry scheduler.
        retryScheduledExecutorService.scheduleWithFixedDelay(new RetryScheduleHandler(retryDataQueue30S),
                retryScheduleInitialDelay, retryScheduleDelay30S, TimeUnit.MILLISECONDS);
        // init 5min retry scheduler.
        retryScheduledExecutorService.scheduleWithFixedDelay(new RetryScheduleHandler(retryDataQueue5Min),
                retryScheduleInitialDelay, retryScheduleDelay5Min, TimeUnit.MILLISECONDS);
        // init 30min retry scheduler.
        retryScheduledExecutorService.scheduleWithFixedDelay(new RetryScheduleHandler(retryDataQueue30Min),
                retryScheduleInitialDelay, retryScheduleDelay30Min, TimeUnit.MILLISECONDS);
    }

    /**
     * 具体业务重试数据处理
     *
     * @param data
     * @return
     */
    protected abstract Integer doRetryData(final List<T> data);

    private final class RetryScheduleHandler implements Runnable {
        /**
         * 定时任务实现
         *
         * @see java.lang.Runnable#run()
         */
        private BlockingQueue<T> dataQueue;

        public RetryScheduleHandler(BlockingQueue<T> retryQueue) {
            dataQueue = retryQueue;
        }

        public void run() {
            List<T> dataList = new ArrayList<T>(retryDataSize);
            while (!dataQueue.isEmpty()) {
                if (dataList.size() >= retryDataSize)
                    break;
                try {
                    T data = dataQueue.take();
                    dataList.add(data);
                } catch (InterruptedException e) {
                    // LogUtil.error(logger, e, "出队列执行异常！");
                }
            }

            Callable<Integer> task = getRetryTask(dataList);
            if (task == null)
                return;
            Future<Integer> future = executorService.submit(task);
            try {
                int result = future.get();
                // LogUtil.info(logger, "任务执行结果：" + result);
            } catch (InterruptedException e) {
                // LogUtil.error(logger, e, "任务执行异常！");
            } catch (ExecutionException e) {
                // LogUtil.error(logger, e, "任务执行异常！");
            }
        }
    }

    /**
     * 创建任务
     *
     * @param data
     * @return
     */
    private Callable<Integer> getRetryTask(final List<T> data) {
        Callable<Integer> task = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                if (data == null) {
                    return 0;
                }
                try {
                    int result = doRetryData(data);
                    return result;
                } catch (Throwable e) {
                    // LogUtil.error(logger, e, "执行任务异常");
                }
                return 0;
            }
        };
        return task;
    }

    /**
     * 添加数据到队列
     *
     * @see com.alipay.csCheck.biz.bench.schedule.ScheduleTask#addData(java.lang.Object)
     */
    @Override
    public void addRetryData(T parameterObject, int retryMode) {
        switch (retryMode) {
        case RETRY_MODE_30S:
            addRetryDateForQueue(parameterObject, retryDataQueue30S);
            break;
        case RETRY_MODE_5MIN:
            addRetryDateForQueue(parameterObject, retryDataQueue5Min);
            break;
        case RETRY_MODE_30MIN:
            addRetryDateForQueue(parameterObject, retryDataQueue30Min);
            break;
        default: // never reach here.
            break;
        }
    }

    /**
     * 按照对应模式添加重试数据到对应队列
     *
     * @param parameterObject
     * @param retryDataQueue
     */
    private void addRetryDateForQueue(T parameterObject, final BlockingQueue<T> retryDataQueue) {
        if (parameterObject != null) {
            if (retryDataQueue.size() >= this.getQueueSize()) {
                // LogUtil.warn(logger, "消费队列数据过大！！！！");

            }
            try {
                retryDataQueue.put(parameterObject);
            } catch (InterruptedException e) {
                // LogUtil.warn(logger, "添加队列数据异常！！！！", e.getMessage());
            }
        }
    }
}

 

package test;

/**
 * 任务定时执行管理器 包含重试任务
 *
 * @author prime7
 */
public interface RetryScheduleTask<T> extends ScheduleTask<T> {

    /**
     * 添加重试数据
     *
     * @param parameterObject
     */
    public void addRetryData(T parameterObject, int retryMode);

}

最后 写个测试类进行一下测试:

 

 

package test;

import java.util.Date;
import java.util.List;

public class TestScheduler extends AbstractRetryScheduleTask<Integer> {

    public static void main(String[] args) {
        TestScheduler test = new TestScheduler();
        test.addData(0);
        test.addRetryData(1, TestScheduler.RETRY_MODE_30S);
        test.addRetryData(2, TestScheduler.RETRY_MODE_5MIN);
        test.addRetryData(3, TestScheduler.RETRY_MODE_30MIN);
        test.addData(0);
        test.addRetryData(1, TestScheduler.RETRY_MODE_30S);
        test.addRetryData(2, TestScheduler.RETRY_MODE_5MIN);
        test.addRetryData(3, TestScheduler.RETRY_MODE_30MIN);
        try {
            System.out.println("now sleep");
            Thread.sleep(10 * 1000);
            System.out.println("now weakup");
            test.addData(0);
            test.addRetryData(1, TestScheduler.RETRY_MODE_30S);
            test.addRetryData(2, TestScheduler.RETRY_MODE_5MIN);
            test.addRetryData(3, TestScheduler.RETRY_MODE_30MIN);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    protected Integer doRetryData(List<Integer> data) {
        return doData(data);
    }

    @Override
    protected Integer doData(List<Integer> data) {
        System.out.println(data + " " + new Date().getTime());
        return 1;
    }
}