java中延迟任务的处理方式

package com.delqueue;  
  
import java.util.concurrent.DelayQueue;  
  
public class Consumer implements Runnable {  
    // 延时队列 ,消费者从其中获取消息进行消费  
    private DelayQueue<Message> queue;  
  
    public Consumer(DelayQueue<Message> queue) {  
        this.queue = queue;  
    }  
  
    @Override  
    public void run() {  
        while (true) {  
            try {  
                Message take = queue.take();  
                System.out.println("消费消息id：" + take.getId() + " 消息体：" + take.getBody());  
            } catch (InterruptedException e) {  
                e.printStackTrace();  
            }  
        }  
    }  
}  

package com.delqueue;  
  
import java.util.concurrent.DelayQueue;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors;  
  
public class DelayQueueTest {  
     public static void main(String[] args) {    
            // 创建延时队列    
            DelayQueue<Message> queue = new DelayQueue<Message>();    
            // 添加延时消息,m1 延时3s    
            Message m1 = new Message(1, "world", 3000);    
            // 添加延时消息,m2 延时10s    
            Message m2 = new Message(2, "hello", 10000);    
            //将延时消息放到延时队列中  
            queue.offer(m2);    
            queue.offer(m1);    
            // 启动消费线程 消费添加到延时队列中的消息，前提是任务到了延期时间   
            ExecutorService exec = Executors.newFixedThreadPool(1);  
            exec.execute(new Consumer(queue));  
            exec.shutdown();  
        }    
}  

package com.test.delayqueue;  
/** 
 * 具体执行相关业务的业务类 
 * @author whd 
 * @date 2017年9月25日 上午12:49:32 
 */  
public class DelayOrderWorker  implements Runnable {  
  
    @Override  
    public void run() {  
        // TODO Auto-generated method stub  
        //相关业务逻辑处理  
        System.out.println(Thread.currentThread().getName()+" do something ……");  
    }  
}  

package com.test.delayqueue;  
  
import java.util.concurrent.Delayed;  
import java.util.concurrent.TimeUnit;  
  
/** 
 * 延时队列中的消息体将任务封装为消息体 
 *  
 * @author whd 
 * @date 2017年9月25日 上午12:48:30 
 * @param <T> 
 */  
public class DelayOrderTask<T extends Runnable> implements Delayed {  
    private final long time;  
    private final T task; // 任务类，也就是之前定义的任务类  
  
    /** 
     * @param timeout 
     *            超时时间(秒) 
     * @param task 
     *            任务 
     */  
    public DelayOrderTask(long timeout, T task) {  
        this.time = System.nanoTime() + timeout;  
        this.task = task;  
    }  
  
    @Override  
    public int compareTo(Delayed o) {  
        // TODO Auto-generated method stub  
        DelayOrderTask other = (DelayOrderTask) o;  
        long diff = time - other.time;  
        if (diff > 0) {  
            return 1;  
        } else if (diff < 0) {  
            return -1;  
        } else {  
            return 0;  
        }  
    }  
  
    @Override  
    public long getDelay(TimeUnit unit) {  
        // TODO Auto-generated method stub  
        return unit.convert(this.time - System.nanoTime(), TimeUnit.NANOSECONDS);  
    }  
  
    @Override  
    public int hashCode() {  
        return task.hashCode();  
    }  
  
    public T getTask() {  
        return task;  
    }  
}  

package com.test.delayqueue;  
  
import java.util.Map;  
import java.util.concurrent.DelayQueue;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors;  
import java.util.concurrent.TimeUnit;  
import java.util.concurrent.atomic.AtomicLong;  
  
/** 
 * 延时队列管理类，用来添加任务、执行任务 
 *  
 * @author whd 
 * @date 2017年9月25日 上午12:44:59 
 */  
public class DelayOrderQueueManager {  
    private final static int DEFAULT_THREAD_NUM = 5;  
    private static int thread_num = DEFAULT_THREAD_NUM;  
    // 固定大小线程池  
    private ExecutorService executor;  
    // 守护线程  
    private Thread daemonThread;  
    // 延时队列  
    private DelayQueue<DelayOrderTask<?>> delayQueue;  
    private static final AtomicLong atomic = new AtomicLong(0);  
    private final long n = 1;  
    private static DelayOrderQueueManager instance = new DelayOrderQueueManager();  
  
    private DelayOrderQueueManager() {  
        executor = Executors.newFixedThreadPool(thread_num);  
        delayQueue = new DelayQueue<>();  
        init();  
    }  
  
    public static DelayOrderQueueManager getInstance() {  
        return instance;  
    }  
  
    /** 
     * 初始化 
     */  
    public void init() {  
        daemonThread = new Thread(() -> {  
            execute();  
        });  
        daemonThread.setName("DelayQueueMonitor");  
        daemonThread.start();  
    }  
  
    private void execute() {  
        while (true) {  
            Map<Thread, StackTraceElement[]> map = Thread.getAllStackTraces();  
            System.out.println("当前存活线程数量:" + map.size());  
            int taskNum = delayQueue.size();  
            System.out.println("当前延时任务数量:" + taskNum);  
            try {  
                // 从延时队列中获取任务  
                DelayOrderTask<?> delayOrderTask = delayQueue.take();  
                if (delayOrderTask != null) {  
                    Runnable task = delayOrderTask.getTask();  
                    if (null == task) {  
                        continue;  
                    }  
                    // 提交到线程池执行task  
                    executor.execute(task);  
                }  
            } catch (Exception e) {  
                e.printStackTrace();  
            }  
        }  
    }  
  
    /** 
     * 添加任务 
     *  
     * @param task 
     * @param time 
     *            延时时间 
     * @param unit 
     *            时间单位 
     */  
    public void put(Runnable task, long time, TimeUnit unit) {  
        // 获取延时时间  
        long timeout = TimeUnit.NANOSECONDS.convert(time, unit);  
        // 将任务封装成实现Delayed接口的消息体  
        DelayOrderTask<?> delayOrder = new DelayOrderTask<>(timeout, task);  
        // 将消息体放到延时队列中  
        delayQueue.put(delayOrder);  
    }  
  
    /** 
     * 删除任务 
     *  
     * @param task 
     * @return 
     */  
    public boolean removeTask(DelayOrderTask task) {  
  
        return delayQueue.remove(task);  
    }  
}