面试题：Java延迟队列与线程池：自定义延迟任务实现

Question

请你实现一个自定义的Java类，使其可以放入`DelayedQueue`中，并且要求该类的延迟时间能够动态调整。同时，说明如何使用线程池来消费这些延迟任务，阐述其中涉及到的线程安全问题及解决方案。

Answer 1

自定义Java类实现Delayed接口

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class MyDelayedTask implements Delayed {
    private long startTime;
    private long delay;
    private String taskName;

    public MyDelayedTask(String taskName, long delay) {
        this.startTime = System.currentTimeMillis();
        this.delay = delay;
        this.taskName = taskName;
    }

    // 动态调整延迟时间
    public void adjustDelay(long newDelay) {
        this.delay = newDelay;
        this.startTime = System.currentTimeMillis();
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long currentTime = System.currentTimeMillis();
        long remainingDelay = startTime + delay - currentTime;
        return unit.convert(remainingDelay, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed other) {
        long diff = this.getDelay(TimeUnit.MILLISECONDS) - other.getDelay(TimeUnit.MILLISECONDS);
        if (diff < 0) {
            return -1;
        } else if (diff > 0) {
            return 1;
        } else {
            return 0;
        }
    }

    @Override
    public String toString() {
        return "MyDelayedTask{" +
                "taskName='" + taskName + '\'' +
                ", delay=" + delay +
                '}';
    }
}

使用线程池消费延迟任务

import java.util.concurrent.*;

public class DelayedTaskConsumer {
    public static void main(String[] args) {
        DelayedQueue<MyDelayedTask> delayedQueue = new DelayedQueue<>();
        ExecutorService executorService = Executors.newFixedThreadPool(1);

        MyDelayedTask task1 = new MyDelayedTask("Task1", 3000);
        MyDelayedTask task2 = new MyDelayedTask("Task2", 5000);

        delayedQueue.add(task1);
        delayedQueue.add(task2);

        executorService.submit(() -> {
            while (true) {
                try {
                    MyDelayedTask task = delayedQueue.take();
                    System.out.println("Processing task: " + task);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });

        // 动态调整任务延迟时间示例
        try {
            TimeUnit.SECONDS.sleep(2);
            task1.adjustDelay(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        executorService.shutdown();
    }
}

线程安全问题及解决方案

1. 线程安全问题：

   • 动态调整延迟时间时，可能存在多个线程同时调用adjustDelay方法，导致数据竞争。
   • 线程池中的线程在消费DelayedQueue中的任务时，DelayedQueue本身需要保证线程安全，否则可能出现不一致的状态。

2. 解决方案：

   • 对于动态调整延迟时间：可以使用synchronized关键字修饰adjustDelay方法，确保同一时间只有一个线程能够修改延迟时间。

   public synchronized void adjustDelay(long newDelay) {
       this.delay = newDelay;
       this.startTime = System.currentTimeMillis();
   }
   

   • 对于DelayedQueue：DelayedQueue本身是线程安全的，它内部使用了锁机制来保证线程安全的操作。在多线程环境下，我们只需要正确使用DelayedQueue的方法（如add、take等），不需要额外的同步操作来保证队列本身的线程安全。