DelayQueue延时队列的使用

女爷i 2022-06-14 08:53 219阅读 0赞

在谈到DelayQueue的使用和原理的时候，我们首先介绍一下DelayQueue，DelayQueue是一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的Delayed 元素。

DelayQueue阻塞队列在我们系统开发中也常常会用到，例如：缓存系统的设计，缓存中的对象，超过了空闲时间，需要从缓存中移出；任务调度系统，能够准确的把握任务的执行时间。我们可能需要通过线程处理很多时间上要求很严格的数据，如果使用普通的线程，我们就需要遍历所有的对象，一个一个的检查看数据是否过期等，首先这样在执行上的效率不会太高，其次就是这种设计的风格也大大的影响了数据的精度。一个需要12:00点执行的任务可能12:01才执行,这样对数据要求很高的系统有更大的弊端。由此我们可以使用DelayQueue。

延迟堵塞队列的实现原理类似于TimerTask。下面假设对超过一定时间的问题的状态修改为实现目标来完成。首先我们利用普通线程对n个问题，然后比对问题的创建时间和现在的时间比较然后来修改问题的状态：

*  我们可以定时的每两分钟查看一次数据对过期的问题的状态进行修改，很显然这种情况对数据的准确性把握不够，其次是不断的查询数据库效率问题。
 *  将问题放在普通的队列中，这种情况需要我们不断的遍历改队列达到修改状态的功能。DelayQueue可以很好的完成这样的工作。

1.首先创建一个Task类，要求实现Delayed，这点类似于TimerTask。设置成员变量“到期时间”,“以及问题对象”并创建构造方法 如下：

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.  **public** **class** Task **implements** Delayed \{  
2.      /\*\* 
3.       \* 到期时间 
4.       \*/  
5.      **private** **final** **long** time;  
6.        
7.      /\*\* 
8.       \* 问题对象 
9.       \*/  
10.     **private** **final** Question question;  
11.     **private** **static** **final** AtomicLong atomic = **new** AtomicLong(0);  
12.       
13.     **private** **final** **long** n;  
14.       
15.       
16.     **public** Task(**long** timeout, Question question) \{  
17.         **this**.time = System.nanoTime() + timeout;  
18.         **this**.question = question;  
19.         **this**.n = atomic.getAndIncrement();  
20.     \}  
21.       
22.     /\*\* 
23.      \* 返回与此对象相关的剩余延迟时间，以给定的时间单位表示 
24.      \*/  
25.     @Override  
26.     **public** **long** getDelay(TimeUnit unit) \{  
27.         **return** unit.convert(**this**.time - System.nanoTime() , TimeUnit.NANOSECONDS);  
28.     \}

2.创建Job类，该类用于执行Task，在这里申明一个线程，该线程用于用队列中取出Task,并将过期的问题状态修改。DelayQueue类的take()方法用于获取并移除此队列的头部，在可从此队列获得到期延迟的元素之前一直等待。如果Task中timeout大于当前的时间则可以通过take()取出到期的对象，如果没有则该队列一直等待。

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.    /\*\* 
2.   \* 守护线程 
3.   \*/  
4.  **private** Thread daemonThread;  
5.    
6.  **public** Job()\{  
7.      Runnable daemonTask = **new** DaemonThread();  
8.      daemonThread = **new** Thread(daemonTask);  
9.      daemonThread.setDaemon(**true**);  
10.     daemonThread.setName("Cache Daemon");  
11.     daemonThread.start();  
12. \}  
13.   
14. //执行线程  
15. **class** DaemonThread **implements** Runnable\{  
16.     @Override  
17.     **public** **void** run() \{  
18.         execute();  
19.     \}     
20. \}  
21.   
22. **public** **void** execute()\{  
23.     System.out.println("start");  
24.     **while**(**true**) \{  
25.         **try** \{  
26.             //从延迟队列中取值,如果没有对象过期则队列一直等待，  
27.             Task t1 = t.take();  
28.             **if** (t1 != **null**) \{  
29.                 //修改问题的状态  
30.                 Question q = t1.getQuestion();  
31.                 System.out.println(q);  
32.                 questionDao.setQuestionSolved(q.getId(), 5600);  
33.             \}  
34.         \} **catch** (Exception e) \{  
35.             e.printStackTrace();  
36.             **break**;  
37.         \}  
38.     \}  
39. \}

3.创建延迟队列将Task加入到延迟队列中，如下：

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.         /\*\* 
2.   \* 创建一个最初为空的新 DelayQueue 
3.   \*/  
4.  **private** DelayQueue t = **new** DelayQueue();  
5.  /\*\* 
6.   \* 添加任务， 
7.   \* time 延迟时间  
8.   \* q 问题 
9.   \* 用户为问题设置延迟时间  
10.  \*/  
11. **public** **void** put(**long** time,Question q)\{  
12.     //转换成ns  
13.     **long** nanoTime = TimeUnit.NANOSECONDS.convert(time, TimeUnit.SECONDS);  
14.     //创建一个任务  
15.     Task k = **new** Task(nanoTime,q);  
16.     //将任务放在延迟的队列中  
17.     t.put(k);  
18. \}

4.测试，检验延迟队列是否可以正常运行。

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.  List<Question> qs = **new** ArrayList<Question>();  
2.      //从数据库中取出10条数据  
3.      **for** (**int** i = 1; i <= 10; i++) \{  
4.      Question q = questionDao.queryQuestionById(i) ;  
5.      qs.add(q);  
6.      \}  
7.      //设置数据库中数据的过期时间  
8.      **for** (**int** k = 1; k <= 10; k++) \{  
9.      j.put(k\*5, qs.get(k - 1));  
10.     \}  
11.     Thread.sleep(200000);

(转载自：http://express.ruanko.com/ruanko-express\_69/tech-overnight5.html)

欢迎加入我的QQ技术交流群425783133

[view plain]: http://blog.csdn.net/gongzi2311/article/details/40397939#