利用DelayQueue实现延时消息队列（简易版MQ）

墨蓝 2022-06-14 08:54 256阅读 0赞

# 1.上文 #

关于阻塞队列的介绍：[http://blog.csdn.net/caicongyang/article/details/50649897][http_blog.csdn.net_caicongyang_article_details_50649897]

# 2.需求 #

延迟消息队列：

1)2个小时后给用户发送短信。  
2)15分钟后关闭网络连接。  
3)2分钟后再次尝试回调。

# 3.案例demo #

Message.[Java][]

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

1.  **package** com.ccy.concurrent;  
2.    
3.  **import** java.util.concurrent.Delayed;  
4.  **import** java.util.concurrent.TimeUnit;  
5.    
6.  /\*\* 
7.   \* <p>  
8.   \* Title: Message.java  
9.   \* Package com.ccy.concurrent  
10.  \* </p> 
11.  \* <p> 
12.  \* Description: 延迟执行的消息 
13.  \* <p> 
14.  \* @author Tom.Cai 
15.  \* @created 2016-2-10 下午7:39:48  
16.  \* @version V1.0  
17.  \* 
18.  \*/  
19. **public** **class** Message **implements** Delayed\{  
20.         **private** String id;  
21.         **private** String name;  
22.         **private** **long** activeTime;//执行时间     
23.           
24.         **public** Message()\{  
25.               
26.         \}  
27.           
28.         **public** Message(String id, String name,**long** activeTime) \{  
29.             **super**();  
30.             **this**.id = id;  
31.             **this**.name = name;  
32.             **this**.activeTime = TimeUnit.NANOSECONDS.convert(activeTime, TimeUnit.MILLISECONDS) + System.nanoTime();  
33.         \}  
34.         **public** String getId() \{  
35.             **return** id;  
36.         \}  
37.         **public** **void** setId(String id) \{  
38.             **this**.id = id;  
39.         \}  
40.         **public** String getName() \{  
41.             **return** name;  
42.         \}  
43.         **public** **void** setName(String name) \{  
44.             **this**.name = name;  
45.         \}  
46.   
47.         @Override  
48.         **public** **int** compareTo(Delayed delayed) \{  
49.             Message msg = (Message)delayed;  
50.             **return** Integer.valueOf(**this**.id)>Integer.valueOf(msg.id)?1:( Integer.valueOf(**this**.id)<Integer.valueOf(msg.id)?-1:0);  
51.         \}  
52.   
53.         @Override  
54.         **public** **long** getDelay(TimeUnit unit) \{  
55.              **return** unit.convert(**this**.activeTime - System.nanoTime(), TimeUnit.NANOSECONDS);   
56.         \}  
57.           
58.           
59.           
60. \}

Producer.javaProducer

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

1.  **package** com.ccy.concurrent;  
2.    
3.  **import** java.util.concurrent.DelayQueue;  
4.    
5.  **public** **class** Producer **implements** Runnable\{  
6.      **private** DelayQueue<Message> queue;  
7.        
8.      **public** Producer(DelayQueue<Message> queue)\{  
9.          **this**.queue = queue;  
10.     \}  
11.   
12.     @Override  
13.     **public** **void** run() \{  
14.         //5秒后发送消息  
15.         Message m2 = **new** Message("2","Tom",5000);  
16.         queue.offer(m2);  
17.         System.out.println("消息生产者往消息队列放置消息："\+m2.getId()+":"\+m2.getName());  
18.         //3秒后发送消息  
19.         Message m1 = **new** Message("1","Tom",3000);  
20.         queue.offer(m1);  
21.         System.out.println("消息生产者往消息队列放置消息："\+m1.getId()+":"\+m1.getName());  
22.           
23.     \}  
24.       
25.   
26. \}

Consumer.java

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

1.  **package** com.ccy.concurrent;  
2.    
3.  **import** java.util.concurrent.DelayQueue;  
4.    
5.  **public** **class** Consumer **implements** Runnable\{  
6.      **private** DelayQueue<Message> queue;  
7.        
8.      **public** Consumer(DelayQueue<Message> queue)\{  
9.          **this**.queue = queue;  
10.     \}  
11.       
12.     @Override  
13.     **public** **void** run() \{  
14.         **while**(**true**)\{  
15.             **try** \{  
16.                 Message take = queue.take();  
17.                 System.out.println("消息需求者获取消息："\+take.getId()+":"\+take.getName());  
18.             \} **catch** (InterruptedException e) \{  
19.                 e.printStackTrace();  
20.             \}  
21.         \}  
22.     \}  
23.   
24. \}

DelayQueueTest.java

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

1.  **package** com.ccy.concurrent;  
2.    
3.  **import** java.util.concurrent.DelayQueue;  
4.    
5.    
6.  **public** **class** DelayQueueTest \{  
7.      **public** **static** **void** main(String\[\] args) \{  
8.          DelayQueue<Message> queue = **new** DelayQueue<Message>();  
9.          **new** Thread(**new** Producer(queue)).start();  
10.         **new** Thread(**new** Consumer(queue)).start();  
11.     \}  
12. \}

# 4.效果 #

![测试效果][Center]

# 5.其他学习资源 #

http://zhangyp.net/rabbitmq-delayqueue/

...

# 6.后记 #

作者将学习Apache的ActiveMQ消息总线，其可以支持定时、延迟投递、重复投递和Cron调度。

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

[http_blog.csdn.net_caicongyang_article_details_50649897]: http://blog.csdn.net/caicongyang/article/details/50649897
[Java]: http://lib.csdn.net/base/java
[view plain]: http://blog.csdn.net/caicongyang/article/details/50650136#
[Center]: /images/20220614/bcb3114fa8f040e0afbbd70ba9478fbc.png