环形队列-高效定时触发

Posted 2023-03-27

参考技术A

为了实现高效延时触发，我们需要实现两个重要的数据结构：

（1）环形队列，例如可以创建一个包含3600个slot的 环形队列 （本质是个数组）

（2）任务集合，环上每一个slot是一个 Set<Task>

同时，启动一个timer，这个timer每隔1s，Current Index指针移动至下一个位置，这个Current Index指针用来标识正在检测的slot。

Task结构中有两个很重要的属性：

（1）Cycle-Num：当Current Index第几圈扫描到这个Slot时，执行任务

（2）TriggerTaskFunction：需要执行的任务指针

假设当前Current Index指向第一格，当有延时任务到达之后，例如希望3610秒之后，触发一个延时任务，只需：

（1） 计算这个Task应该放在哪一个slot ，现在指向1，3610秒之后，应该是第11格，所以这个Task应该放在第11个slot的Set<Task>中

（2） 计算这个Task的Cycle-Num ，由于环形队列是3600格（每秒移动一格，正好1小时），这个任务是3610秒后执行，所以应该绕3610/3600=1圈之后再执行，于是Cycle-Num=1

Current Index不停的移动，每秒移动到一个新slot，这个slot中对应的Set<Task>，每个Task看Cycle-Num是不是0：

（1）如果不是0，说明还需要多移动几圈，将Cycle-Num减1

（2）如果是0，说明马上要执行这个Task了，取出TriggerTaskFunction执行（可以用单独的线程来执行Task），并把这个Task从Set<Task>中删除

环形队列高效触发大量超时任务的算法实现

基于环形队列的超时触发算法只需要一个timer即可实现批量超时任务的触发，CPU消耗低，效率高。下面是此算法的简单实现。

1，TaskHolder.java

package com.zws.timer;
/**
 * 
 * @author wensh.zhu
 * @date 2018-04-22
 */
public class TaskHolder {
	
	/** 任务所需等待的圈数，即任务需要走几圈**/
	private int cycles;
	private int delays;
	private Runnable task;
	
	public TaskHolder() {}

	public TaskHolder(int cycles, int delays, Runnable task) {
		this.cycles = cycles;
		this.delays = delays;
		this.task = task;
	}
	
	public boolean isTimeOut() {
		return cycles <= 0;
	}
	
	public void cutDown() {
		cycles --;
	}

	public int getCycles() {
		return cycles;
	}

	public void setCycles(int cycles) {
		this.cycles = cycles;
	}

	public int getDelays() {
		return delays;
	}

	public void setDelays(int delays) {
		this.delays = delays;
	}

	public Runnable getTask() {
		return task;
	}

	public void setTask(Runnable task) {
		this.task = task;
	}

	@Override
	public String toString() {
		return "TaskHolder[cycles=" + cycles + ", delays=" + delays + "]";
	}
	
}

2，TimerContext.java

package com.zws.timer;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
 * 
 * @author wensh.zhu
 * @date 2018-04-22
 */
public class TimerContext {

	public static final int DEFAULT_TICKS = 60;
	public static final int DEFAULT_TICK_DURATION = 1;
	
	private Map<Integer, Queue<TaskHolder>> taskHolders;
	private volatile int currentTick = 0;
	
	/** tick一圈的长度 **/
	private int ticks = DEFAULT_TICKS;
	
	/** 每tick一次的时间间隔，单位：秒**/
	private int tickDuration = DEFAULT_TICK_DURATION;
	
	public TimerContext() {
		init();
	}

	public TimerContext(int ticks, int tickDuration) {
		if (ticks <= 0)
			throw new IllegalArgumentException("ticks must be greater than 0");
		
		if (tickDuration <= 0)
			throw new IllegalArgumentException("tickDuration must be greater than 0");
		
		this.ticks = ticks;
		this.tickDuration = tickDuration;
		init();
	}
	
	private void init() {
		taskHolders = new ConcurrentHashMap<Integer, Queue<TaskHolder>>();
		for (int i = 0; i < ticks; i ++)
			taskHolders.put(i, new ConcurrentLinkedQueue<TaskHolder>());
	}
	
	/**
	 * 添加一个定时任务并计算需要走的圈数和落脚的index
	 * @param task
	 * @param delays
	 */
	public void addTask(Runnable task, int delays) {
		if (task == null) 
			throw new NullPointerException("task must not be null");
		
		if (delays <=0) 
			throw new IllegalArgumentException("delays must be greater than 0");
		
		int allSeconds = ticks * tickDuration;
		int cycles = delays / allSeconds;
		int index = ((delays % allSeconds) / tickDuration) + currentTick;
		TaskHolder metaData = new TaskHolder(cycles, delays, task);
		taskHolders.get(index).add(metaData);
	}
	
	public int tick() {
		currentTick = (currentTick + 1) % ticks;
		return currentTick;
	}
	
	public Queue<TaskHolder> getCurrentTasks() {
		return taskHolders.get(currentTick);
	}

	public int getCurrentTick() {
		return currentTick;
	}

	public int getTicks() {
		return ticks;
	}

	public int getTickDuration() {
		return tickDuration;
	}
	
	@Override
	public String toString() {
		return "TimerContext [timers=" + taskHolders + ", ticks=" + ticks + ", tickDuration=" + tickDuration
				+ ", currentTick=" + currentTick + "]";
	}
}

3，TimerScheduler.java

package com.zws.timer;

import java.io.IOException;
import java.util.Iterator;
import java.util.Queue;
import java.util.Timer;
import java.util.TimerTask;
/**
 * 用于判断定时器是否到时、执行任务、维护定时器状态。
 * @author wensh.zhu
 * @date 2018-04-22
 */
public class TimerScheduler extends TimerTask {
	
	private TimerContext timerContext;
	
	public TimerScheduler() {}

	public TimerScheduler(TimerContext timerContext) {
		this.timerContext = timerContext;
	}
	
	/**
	 * 定时检测，如果定时器触发时间到了就从集合中删除并执行任务，否则圈数减一。
	 */
	@Override
	public void run() {
		if (timerContext == null) 
			return;
		
		Queue<TaskHolder> timers = timerContext.getCurrentTasks();
		Iterator<TaskHolder> itor = timers.iterator();
		while (itor.hasNext()) {
			TaskHolder timer = itor.next();
			if (timer.isTimeOut()) {
				itor.remove();
				new Thread(timer.getTask()).start();
			} else {
				timer.cutDown();
			}
		}
		
		timerContext.tick();
	}
	
	public void addTask(Runnable task, int delays) {
		timerContext.addTask(task, delays);
	}

	public TimerContext getTimerContext() {
		return timerContext;
	}

	public void setTimerContext(TimerContext timerContext) {
		this.timerContext = timerContext;
	}
	
	public static void main(String[] args) throws IOException {
		TimerContext context = new TimerContext(60, 1);
		TimerScheduler sheduler = new TimerScheduler(context);
		sheduler.addTask(new Runnable() {
			
			public void run() {
				System.out.println(DateUtils.now());
			}
		}, 60);
		System.out.println(DateUtils.now());
		
		Timer timer = new Timer();
		timer.scheduleAtFixedRate(sheduler, 0, context.getTickDuration() * 1000L);
		
		System.in.read();
	}
	
}

4，DateUtils.java

package com.zws.timer;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
/**
 * 
 * @author wensh.zhu
 * @date 2018-04-22
 */
public class DateUtils {
	public static final String DEFAULT_PATTERN = "yyyy-MM-dd HH:mm:ss";

	public static String now() {
		LocalDateTime time = LocalDateTime.now();
		return time.format(DateTimeFormatter.ofPattern(DEFAULT_PATTERN));
	}
	
	public static String plusSeconds(int seconds) {
		LocalDateTime time = LocalDateTime.now();
		time.plusSeconds(seconds);
		return time.format(DateTimeFormatter.ofPattern(DEFAULT_PATTERN));
	}
}