scheduler.java

短信开发用于文件交换处理转发的类模块

package com.pub.timeswan;

/**
 * <p>Title: </p>
 *
 * <p>Description: </p>
 *
 * <p>Copyright: Copyright (c) 2006</p>
 *
 * <p>Company: </p>
 *
 * @author not attributable
 * @version 1.0
 */

import java.util.*;
import java.util.concurrent.*;
import org.apache.log4j.*;
public class Scheduler extends Thread {

    private Logger log = Logger.getLogger(Scheduler.class);

    public Scheduler(int interval) {
        _runCtrl = true;
        _loopInterval = interval;
    }

    private ArrayList _taskWrappers = new ArrayList(20);
    ThreadPoolExecutor _pool = new ThreadPoolExecutor(1, 10, 60,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue(5),
            new ThreadPoolExecutor.DiscardPolicy());
//    ThreadPoolExecutor _pool = new Executors

    public synchronized int addTask(TaskWrapper tw) {
        _taskWrappers.add(tw);
        return 0;
    }

//    synchronized int delTask(TaskWrapper tw) {
//        _taskWrappers.remove(tw);
//        return 0;
//    }

    synchronized void clearTask() {
        if (_taskWrappers != null) {
            for (int i = 0; i < _taskWrappers.size(); i++) {
                if (((TaskWrapper) _taskWrappers.get(i)).state() ==
                    TaskWrapper.INVALID) {
                    _taskWrappers.remove(i--);
                }
            }
        }
    }

    private boolean _runCtrl = true;
    private long _loopInterval = 200;

    public synchronized void stopthread() {
        _runCtrl = false;
        _pool.shutdown();
        clearTask();
    }

    public void run() {

        long start = 0;
        long runTime, diff;

        while (_runCtrl) {

            //本次循环开始时间
            start = System.currentTimeMillis();

            clearTask();

            synchronized (this) {
                for (Object obj : _taskWrappers) {

                    TaskWrapper tw = (TaskWrapper) obj;

                    //维护运行对象列表，检查是否可以并发运行
                    if (tw.countRunning() > 0 && !tw.canCocurrent()) {
                        continue;
                    }
                    if (tw.state() == tw.CANCELLED || tw.state() == tw.INVALID) {
                        continue;
                    }

                    try {
                        runTime = CrontabExpr.Parse(tw.expr()).getTimeInMillis();
                    } catch (Throwable ex) {
                        log.error("parse CrontabExpr error:" + tw.expr(), ex);
                        continue;
                    }

                    diff = start - runTime;

                    if (diff >= 0 && diff < _loopInterval) {

                        ITask task = tw.getTaskObject();
                        if (tw.newInstance()) {
                            try {
                                task = (ITask) tw.getTaskClass().newInstance();
                            }
                            /*catch (IllegalAccessException ex2) {
                             } catch (InstantiationException ex2) {
                                                     }*/
                            catch (Throwable ex) {
                                log.error(null, ex);
                                task = null;
                            }
                            //if(task != null) task.run();
                        }
                        if (task != null) {
//                            log.debug(task+" "+tw.expr() + "  " +
//                                  new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").
//                                  format(new Date(runTime)));

                            tw.addRunning(task);
                            _pool.execute(task);
                            Thread.yield();
                            tw.startTime(System.currentTimeMillis());
                            tw.incTimesRun();
                        }
                    }
                }
            }
            diff = System.currentTimeMillis() - start;

            try {
                Thread.sleep(Math.max(0, _loopInterval - diff));
            } catch (InterruptedException ex1) {
                _runCtrl = false;
            } catch (Throwable ex) {
                log.error(null, ex);
            }
        }
    }

    public void runCtrl(boolean _runCtrl) {
        this._runCtrl = _runCtrl;
    }
}