定时任务实现原理详解

publicstaticvoidmain(String[]args){
finallongtimeInterval=5000;
newThread(newRunnable(){
@Override
publicvoidrun(){
while(true){
System.out.println(Thread.currentThread().getName()+"每隔5秒执行一次");
try{
Thread.sleep(timeInterval);
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}
}).start();
}

publicstaticvoidmain(String[]args){
Timertimer=newTimer();
//每隔1秒调用一次
timer.schedule(newTimerTask(){
@Override
publicvoidrun(){
System.out.println("test1");
}
},1000,1000);
//每隔3秒调用一次
timer.schedule(newTimerTask(){
@Override
publicvoidrun(){
System.out.println("test2");
}
},3000,3000);
}

privatevoidsched(TimerTasktask,longtime,longperiod){
if(time<0)
thrownewIllegalArgumentException("Illegal execution time.");
// Constrain value of period sufficiently to prevent numeric
// overflow while still being effectively infinitely large.
if(Math.abs(period)>(Long.MAX_VALUE>>1))
period>>=1;
synchronized(queue){
if(!thread.newTasksMayBeScheduled)
thrownewIllegalStateException("Timer already cancelled.");
synchronized(task.lock){
if(task.state!=TimerTask.VIRGIN)
thrownewIllegalStateException(
"Task already scheduled or cancelled");
task.nextExecutionTime=time;
task.period=period;
task.state=TimerTask.SCHEDULED;
}
queue.add(task);
if(queue.getMin()==task)
queue.notify();
}
}

publicclassTimer{
privatefinalTaskQueuequeue=newTaskQueue();
privatefinalTimerThreadthread=newTimerThread(queue);
publicTimer(){
this("Timer-"+serialNumber());
}
publicTimer(Stringname){
thread.setName(name);
thread.start();
}
//...
}

classTaskQueue{
privateTimerTask[]queue=newTimerTask[128];
privateintsize=0;
voidadd(TimerTasktask){
// Grow backing store if necessary
if(size+1==queue.length)
queue=Arrays.copyOf(queue,2*queue.length);
queue[++size]=task;
fixUp(size);
}
privatevoidfixUp(intk){
while(k>1){
intj=k>>1;
if(queue[j].nextExecutionTime<=queue[k].nextExecutionTime)
break;
TimerTasktmp=queue[j];
queue[j]=queue[k];
queue[k]=tmp;
k=j;
}
}
//....
}

classTimerThreadextendsThread{
booleannewTasksMayBeScheduled=true;
privateTaskQueuequeue;
TimerThread(TaskQueuequeue){
this.queue=queue;
}
publicvoidrun(){
try{
mainLoop();
}finally{
// Someone killed this Thread, behave as if Timer cancelled
synchronized(queue){
newTasksMayBeScheduled=false;
queue.clear();// Eliminate obsolete references
}
}
}
/**
* The main timer loop.  (See class comment.)
*/
privatevoidmainLoop(){
while(true){
try{
TimerTasktask;
booleantaskFired;
synchronized(queue){
// Wait for queue to become non-empty
while(queue.isEmpty()&&newTasksMayBeScheduled)
queue.wait();
if(queue.isEmpty())
break;// Queue is empty and will forever remain; die
// Queue nonempty; look at first evt and do the right thing
longcurrentTime,executionTime;
task=queue.getMin();
synchronized(task.lock){
if(task.state==TimerTask.CANCELLED){
queue.removeMin();
continue;// No action required, poll queue again
}
currentTime=System.currentTimeMillis();
executionTime=task.nextExecutionTime;
if(taskFired=(executionTime<=currentTime)){
if(task.period==0){// Non-repeating, remove
queue.removeMin();
task.state=TimerTask.EXECUTED;
}else{// Repeating task, reschedule
queue.rescheduleMin(
task.period<0?currentTime-task.period
:executionTime+task.period);
}
}
}
if(!taskFired)// Task hasn't yet fired; wait
queue.wait(executionTime-currentTime);
}
if(taskFired)// Task fired; run it, holding no locks
task.run();
}catch(InterruptedExceptione){
}
}
}
}

publicstaticvoidmain(String[]args){
ScheduledThreadPoolExecutorexecutor=newScheduledThreadPoolExecutor(3);
//启动1秒之后，每隔1秒执行一次
executor.scheduleAtFixedRate((newRunnable(){
@Override
publicvoidrun(){
System.out.println("test3");
}
}),1,1,TimeUnit.SECONDS);
//启动1秒之后，每隔3秒执行一次
executor.scheduleAtFixedRate((newRunnable(){
@Override
publicvoidrun(){
System.out.println("test4");
}
}),1,3,TimeUnit.SECONDS);
}

publicScheduledFuture<?>scheduleAtFixedRate(Runnablecommand,
longinitialDelay,
longperiod,
TimeUnitunit){
if(command==null||unit==null)
thrownewNullPointerException();
if(period<=0)
thrownewIllegalArgumentException();
ScheduledFutureTask<Void>sft=
newScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay,unit),
unit.toNanos(period));
RunnableScheduledFuture<Void>t=decorateTask(command,sft);
sft.outerTask=t;
delayedExecute(t);
returnt;
}

privatevoiddelayedExecute(RunnableScheduledFuture<?>task){
if(isShutdown())
reject(task);
else{
super.getQueue().add(task);
if(isShutdown()&&
!canRunInCurrentRunState(task.isPeriodic())&&
remove(task))
task.cancel(false);
else
//预处理
ensurePrestart();
}
}

publicScheduledThreadPoolExecutor(intcorePoolSize){
super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,
newDelayedWorkQueue());
}

staticclassDelayedWorkQueueextendsAbstractQueue<Runnable>
implementsBlockingQueue<Runnable>{
privatestaticfinalintINITIAL_CAPACITY=16;
privateRunnableScheduledFuture<?>[]queue=
newRunnableScheduledFuture<?>[INITIAL_CAPACITY];
privatefinalReentrantLocklock=newReentrantLock();
privateintsize=0;
//....
publicbooleanadd(Runnablee){
returnoffer(e);
}
publicbooleanoffer(Runnablex){
if(x==null)
thrownewNullPointerException();
RunnableScheduledFuture<?>e=(RunnableScheduledFuture<?>)x;
finalReentrantLocklock=this.lock;
lock.lock();
try{
inti=size;
if(i>=queue.length)
grow();
size=i+1;
if(i==0){
queue[0]=e;
setIndex(e,0);
}else{
siftUp(i,e);
}
if(queue[0]==e){
leader=null;
available.signal();
}
}finally{
lock.unlock();
}
returntrue;
}
publicRunnableScheduledFuture<?>take()throwsInterruptedException{
finalReentrantLocklock=this.lock;
lock.lockInterruptibly();
try{
for(;;){
RunnableScheduledFuture<?>first=queue[0];
if(first==null)
available.await();
else{
longdelay=first.getDelay(NANOSECONDS);
if(delay<=0)
returnfinishPoll(first);
first=null;// don't retain ref while waiting
if(leader!=null)
available.await();
else{
ThreadthisThread=Thread.currentThread();
leader=thisThread;
try{
available.awaitNanos(delay);
}finally{
if(leader==thisThread)
leader=null;
}
}
}
}
}finally{
if(leader==null&&queue[0]!=null)
available.signal();
lock.unlock();
}
}
}

privateclassScheduledFutureTask<V>
extendsFutureTask<V>implementsRunnableScheduledFuture<V>{
/** Sequence number to break ties FIFO */
privatefinallongsequenceNumber;
/** The time the task is enabled to execute in nanoTime units */
privatelongtime;
/**
* Period in nanoseconds for repeating tasks.  A positive
* value indicates fixed-rate execution.  A negative value
* indicates fixed-delay execution.  A value of 0 indicates a
* non-repeating task.
*/
privatefinallongperiod;
/** The actual task to be re-enqueued by reExecutePeriodic */
RunnableScheduledFuture<V>outerTask=this;
/**
* Overrides FutureTask version so as to reset/requeue if periodic.
*/
publicvoidrun(){
booleanperiodic=isPeriodic();
if(!canRunInCurrentRunState(periodic))
cancel(false);
elseif(!periodic)
ScheduledFutureTask.super.run();
elseif(ScheduledFutureTask.super.runAndReset()){
setNextRunTime();
reExecutePeriodic(outerTask);
}
}
//...
}

publicclassRingBufferWheel{
privateLoggerlogger=LoggerFactory.getLogger(RingBufferWheel.class);
/**
* default ring buffer size
*/
privatestaticfinalintSTATIC_RING_SIZE=64;
privateObject[]ringBuffer;
privateintbufferSize;
/**
* business thread pool
*/
privateExecutorServiceexecutorService;
privatevolatileintsize=0;
/***
* task stop sign
*/
privatevolatilebooleanstop=false;
/**
* task start sign
*/
privatevolatileAtomicBooleanstart=newAtomicBoolean(false);
/**
* total tick times
*/
privateAtomicIntegertick=newAtomicInteger();
privateLocklock=newReentrantLock();
privateConditioncondition=lock.newCondition();
privateAtomicIntegertaskId=newAtomicInteger();
privateMap<Integer,Task>taskMap=newConcurrentHashMap<>(16);
/**
* Create a new delay task ring buffer by default size
*
* @param executorService the business thread pool
*/
publicRingBufferWheel(ExecutorServiceexecutorService){
this.executorService=executorService;
this.bufferSize=STATIC_RING_SIZE;
this.ringBuffer=newObject[bufferSize];
}
/**
* Create a new delay task ring buffer by custom buffer size
*
* @param executorService the business thread pool
* @param bufferSize      custom buffer size
*/
publicRingBufferWheel(ExecutorServiceexecutorService,intbufferSize){
this(executorService);
if(!powerOf2(bufferSize)){
thrownewRuntimeException("bufferSize=["+bufferSize+"] must be a power of 2");
}
this.bufferSize=bufferSize;
this.ringBuffer=newObject[bufferSize];
}
/**
* Add a task into the ring buffer(thread safe)
*
* @param task business task extends {@link Task}
*/
publicintaddTask(Tasktask){
intkey=task.getKey();
intid;
try{
lock.lock();
intindex=mod(key,bufferSize);
task.setIndex(index);
Set<Task>tasks=get(index);
intcycleNum=cycleNum(key,bufferSize);
if(tasks!=null){
task.setCycleNum(cycleNum);
tasks.add(task);
}else{
task.setIndex(index);
task.setCycleNum(cycleNum);
Set<Task>sets=newHashSet<>();
sets.add(task);
put(key,sets);
}
id=taskId.incrementAndGet();
task.setTaskId(id);
taskMap.put(id,task);
size++;
}finally{
lock.unlock();
}
start();
returnid;
}
/**
* Cancel task by taskId
* @param id unique id through {@link #addTask(Task)}
* @return
*/
publicbooleancancel(intid){
booleanflag=false;
Set<Task>tempTask=newHashSet<>();
try{
lock.lock();
Tasktask=taskMap.get(id);
if(task==null){
returnfalse;
}
Set<Task>tasks=get(task.getIndex());
for(Tasktk:tasks){
if(tk.getKey()==task.getKey()&&tk.getCycleNum()==task.getCycleNum()){
size--;
flag=true;
taskMap.remove(id);
}else{
tempTask.add(tk);
}
}
//update origin data
ringBuffer[task.getIndex()]=tempTask;
}finally{
lock.unlock();
}
returnflag;
}
/**
* Thread safe
*
* @return the size of ring buffer
*/
publicinttaskSize(){
returnsize;
}
/**
* Same with method {@link #taskSize}
* @return
*/
publicinttaskMapSize(){
returntaskMap.size();
}
/**
* Start background thread to consumer wheel timer, it will always run until you call method {@link #stop}
*/
publicvoidstart(){
if(!start.get()){
if(start.compareAndSet(start.get(),true)){
logger.info("Delay task is starting");
Threadjob=newThread(newTriggerJob());
job.setName("consumer RingBuffer thread");
job.start();
start.set(true);
}
}
}
/**
* Stop consumer ring buffer thread
*
* @param force True will force close consumer thread and discard all pending tasks
*              otherwise the consumer thread waits for all tasks to completes before closing.
*/
publicvoidstop(booleanforce){
if(force){
logger.info("Delay task is forced stop");
stop=true;
executorService.shutdownNow();
}else{
logger.info("Delay task is stopping");
if(taskSize()>0){
try{
lock.lock();
condition.await();
stop=true;
}catch(InterruptedExceptione){
logger.error("InterruptedException",e);
}finally{
lock.unlock();
}
}
executorService.shutdown();
}
}
privateSet<Task>get(intindex){
return(Set<Task>)ringBuffer[index];
}
privatevoidput(intkey,Set<Task>tasks){
intindex=mod(key,bufferSize);
ringBuffer[index]=tasks;
}
/**
* Remove and get task list.
* @param key
* @return task list
*/
privateSet<Task>remove(intkey){
Set<Task>tempTask=newHashSet<>();
Set<Task>result=newHashSet<>();
Set<Task>tasks=(Set<Task>)ringBuffer[key];
if(tasks==null){
returnresult;
}
for(Tasktask:tasks){
if(task.getCycleNum()==0){
result.add(task);
size2Notify();
}else{
// decrement 1 cycle number and update origin data
task.setCycleNum(task.getCycleNum()-1);
tempTask.add(task);
}
// remove task, and free the memory.
taskMap.remove(task.getTaskId());
}
//update origin data
ringBuffer[key]=tempTask;
returnresult;
}
privatevoidsize2Notify(){
try{
lock.lock();
size--;
if(size==0){
condition.signal();
}
}finally{
lock.unlock();
}
}
privatebooleanpowerOf2(inttarget){
if(target<0){
returnfalse;
}
intvalue=target&(target-1);
if(value!=0){
returnfalse;
}
returntrue;
}
privateintmod(inttarget,intmod){
// equals target % mod
target=target+tick.get();
returntarget&(mod-1);
}
privateintcycleNum(inttarget,intmod){
//equals target/mod
returntarget>>Integer.bitCount(mod-1);
}
/**
* An abstract class used to implement business.
*/
publicabstractstaticclassTaskextendsThread{
privateintindex;
privateintcycleNum;
privateintkey;
/**
* The unique ID of the task
*/
privateinttaskId;
@Override
publicvoidrun(){
}
publicintgetKey(){
returnkey;
}
/**
*
* @param key Delay time(seconds)
*/
publicvoidsetKey(intkey){
this.key=key;
}
publicintgetCycleNum(){
returncycleNum;
}
privatevoidsetCycleNum(intcycleNum){
this.cycleNum=cycleNum;
}
publicintgetIndex(){
returnindex;
}
privatevoidsetIndex(intindex){
this.index=index;
}
publicintgetTaskId(){
returntaskId;
}
publicvoidsetTaskId(inttaskId){
this.taskId=taskId;
}
}
privateclassTriggerJobimplementsRunnable{
@Override
publicvoidrun(){
intindex=0;
while(!stop){
try{
Set<Task>tasks=remove(index);
for(Tasktask:tasks){
executorService.submit(task);
}
if(++index>bufferSize-1){
index=0;
}
//Total tick number of records
tick.incrementAndGet();
TimeUnit.SECONDS.sleep(1);
}catch(Exceptione){
logger.error("Exception",e);
}
}
logger.info("Delay task has stopped");
}
}
}