Java：自定义ScheduledExecutorService来实现定时提交任务

Posted 2021-07-24 你是小KS

1. 声明

当前内容主要为本人学习和复习定时任务

一般的定时任务：

1. 使用Executors.newScheduledThreadPool方式创建
2. 调用scheduleWithFixedDelay来定时执行

主要为实现定时周期性任务，但是每个周期性的任务执行的时间不确定，这个就导致一些问题例如：必须每天提交日志信息(不能延迟)

2.newScheduledThreadPool的demo

如下：一个1秒钟的定时任务

public static void main(String[] args) {
		ScheduledExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(3);
		DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
		Runnable run = () -> {
			String name = Thread.currentThread().getName();
			System.out.println(name + ":正在执行任务！" + format.format(new Date()));
			try {
				Thread.sleep(4000l);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			System.out.println(name + ":执行完毕！" + format.format(new Date()));
		};

		try {

			/* newScheduledThreadPool.schedule(run, 1000L, TimeUnit.MILLISECONDS); */

			newScheduledThreadPool.scheduleWithFixedDelay(run, 0L, 1000L, TimeUnit.MILLISECONDS);
			System.out.println("执行任务完毕了吗？");
		} catch (Exception ex) {
			ex.printStackTrace();
		}

	}

但是实际任务在runnable中模拟执行时间为4秒

执行结果为：

无论将初始化的poolSize设置为多少，那么永远只有一个在执行，并且执行间隔为上一个任务执行完毕后+1秒(这个有的时候并不符合我们的任务要求)

就像现实生活中，任务并不会等待后再执行，而是继续提交执行

3. 自定的实现方式定时提交任务

1.任务必须定时提交，不关心执行任务的完成时间(可以考虑使用定时器+线程池方式)

public class MySchaduleThread implements ScheduledExecutorService {
	private final int poolMaxSize;
	private final ScheduledExecutorService scheduledExecutorService; // 控制定时任务
	private final ExecutorService executorService;// 控制执行任务，按照时间间隔将定时任务交给这个进行处理
	
	// 缺少线程状态的记录,缺少已经正在执行的任务线程数量记录
	// 缺少日志信息

	public MySchaduleThread(int poolMaxSize) {
		this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
		this.poolMaxSize = poolMaxSize;
		this.executorService = Executors.newFixedThreadPool(poolMaxSize);
	}

	public static void main(String[] args) {
			MySchaduleThread schaduleThread=new MySchaduleThread(5);
			DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
			Runnable run = () -> {
				String name = Thread.currentThread().getName();
				System.out.println(name + ":正在执行任务！" + format.format(new Date()));
				try {
					Thread.sleep(4000l);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				System.out.println(name + ":执行完毕！" + format.format(new Date()));
			};
			schaduleThread.scheduleWithFixedDelay(run, 0, 1000L, TimeUnit.MILLISECONDS);
		
	}

	@Override
	public void shutdown() {
		executorService.shutdown();
		scheduledExecutorService.shutdown();
	}

	@Override
	public List<Runnable> shutdownNow() {
		scheduledExecutorService.shutdownNow();
		List<Runnable> shutdownNow = executorService.shutdownNow();
		return shutdownNow;
	}

	@Override
	public boolean isShutdown() {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public boolean isTerminated() {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public <T> Future<T> submit(Callable<T> task) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <T> Future<T> submit(Runnable task, T result) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Future<?> submit(Runnable task) {
		// TODO Auto-generated method stub
		return executorService.submit(task);
	}

	@Override
	public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
			throws InterruptedException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
			throws InterruptedException, ExecutionException, TimeoutException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public void execute(Runnable command) {
		// TODO Auto-generated method stub

	}

	@Override
	public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
		return scheduledExecutorService.scheduleWithFixedDelay(() -> {
			submit(command);
		}, initialDelay, delay, unit);
	}

}

执行结果：

测试是成功

4. 思考

1. 由于上面的是最简单的版本实现的，但是可能存在所有执行的线程都是繁忙，出现最后线程不处理的情况，需要处理其他的问题

2. 线程必须提交所有执行任务的状态，可以实现线程的任务的自由取消，执行超时等

3. 线程必须有最大执行数量控制，各种错误处理等