精巧好用的DelayQueue

Posted 2020-09-26 薇薇一笑g

我们谈一下实际的场景吧。我们在开发中，有如下场景

a) 关闭空闲连接。服务器中，有很多客户端的连接，空闲一段时间之后需要关闭之。
b) 缓存。缓存中的对象，超过了空闲时间，需要从缓存中移出。
c) 任务超时处理。在网络协议滑动窗口请求应答式交互时，处理超时未响应的请求。

一种笨笨的办法就是，使用一个后台线程，遍历所有对象，挨个检查。这种笨笨的办法简单好用，但是对象数量过多时，可能存在性能问题，检查间隔时间不好设置，间隔时间过大，影响精确度，多小则存在效率问题。而且做不到按超时的时间顺序处理。 

这场景，使用DelayQueue最适合了。

DelayQueue是java.util.concurrent中提供的一个很有意思的类。很巧妙，非常棒！但是java doc和Java SE 5.0的source中都没有提供Sample。我最初在阅读ScheduledThreadPoolExecutor源码时，发现DelayQueue的妙用。随后在实际工作中，应用在session超时管理，网络应答通讯协议的请求超时处理。

本文将会对DelayQueue做一个介绍，然后列举应用场景。并且提供一个Delayed接口的实现和Sample代码。

DelayQueue是一个BlockingQueue，其特化的参数是Delayed。（不了解BlockingQueue的同学，先去了解BlockingQueue再看本文）
Delayed扩展了Comparable接口，比较的基准为延时的时间值，Delayed接口的实现类getDelay的返回值应为固定值（final）。DelayQueue内部是使用PriorityQueue实现的。

DelayQueue = BlockingQueue + PriorityQueue + Delayed

DelayQueue的关键元素BlockingQueue、PriorityQueue、Delayed。可以这么说，DelayQueue是一个使用优先队列（PriorityQueue）实现的BlockingQueue，优先队列的比较基准值是时间。

他们的基本定义如下

public interface Comparable<T> {
    public int compareTo(T o);
}

public interface Delayed extends Comparable<Delayed> {
    long getDelay(TimeUnit unit);
}

public class DelayQueue<E extends Delayed> implements BlockingQueue<E> { 
    private final PriorityQueue<E> q = new PriorityQueue<E>();
}

转自：http://www.cnblogs.com/jobs/archive/2007/04/27/730255.html

 

 

模拟一个考试的日子，考试时间为120分钟，30分钟后才可交卷，当时间到了，或学生都交完卷了考试结束。

这个场景中几个点需要注意：

1. 考试时间为120分钟，30分钟后才可交卷，初始化考生完成试卷时间最小应为30分钟
2. 对于能够在120分钟内交卷的考生，如何实现这些考生交卷
3. 对于120分钟内没有完成考试的考生，在120分钟考试时间到后需要让他们强制交卷
4. 在所有的考生都交完卷后，需要将控制线程关闭

抽象出两个类，学生类和老师类，用DelayQueue存储考生（Student类）。每一个考生都有自己的名字和完成试卷的时间

Teacher线程对DelayQueue进行监控，收取完成试卷小于120分钟的学生的试卷。当考试时间120分钟到时，teacher线程宣布考试结束，强制DelayQueue中还存在的考生交卷。

package zhongqiu.common.base.thread;
 
import java.util.Iterator;
import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
 
public class DelayQueueDemoOne {
 
    public static void main(String[] args) throws InterruptedException {
        // TODO Auto-generated method stub
        int studentNumber = 20;
        DelayQueue<Student> students = new DelayQueue<Student>();
        Random random = new Random();
        for (int i = 0; i < studentNumber; i++) {
            students.put(new Student("student" + (i + 1), 30 + random.nextInt(120)));
        }
        students.put(new Student("student",120));
        Thread teacherThread = new Thread(new Teacher(students));
        teacherThread.start();
    }
}
 
class Student implements Runnable, Delayed {
 
    private String name;
    public long workTime;
    private long submitTime;
    private boolean isForce = false;
 
    public Student() {
    }
 
    public Student(String name, long workTime) {
        this.name = name;
        this.workTime = workTime;
        this.submitTime = TimeUnit.NANOSECONDS.convert(workTime, TimeUnit.NANOSECONDS) + System.nanoTime();// 纳秒级别
    }
 
    @Override
    public int compareTo(Delayed o) {
        // TODO Auto-generated method stub
        if (o == null || !(o instanceof Student))
            return 1;
        if (o == this)
            return 0;
        Student s = (Student) o;
        if (this.workTime > s.workTime) {
            return 1;
        } else if (this.workTime == s.workTime) {
            return 0;
        } else {
            return -1;
        }
    }
 
    @Override
    public long getDelay(TimeUnit unit) {
        // TODO Auto-generated method stub
        return unit.convert(submitTime - System.nanoTime(), TimeUnit.NANOSECONDS);
    }
 
    @Override
    public void run() {
        // TODO Auto-generated method stub
        if (isForce) {
            System.out.println(name + " 交卷，实际用时 120分钟");
        } else {
            System.out.println(name + " 交卷," + "实际用时 " + workTime + " 分钟");
        }
    }
 
    public boolean isForce() {
        return isForce;
    }
 
    public void setForce(boolean isForce) {
        this.isForce = isForce;
    }
 
}
 
class Teacher implements Runnable {
    private int counter = 20;
    private DelayQueue<Student> students;
 
    public Teacher(DelayQueue<Student> students) {
        this.students = students;
    }
 
    @Override
    public void run() {
        // TODO Auto-generated method stub
        try {
            System.out.println(" test start");
            while (counter > 0) {
                Student student = students.poll();
                if (student.workTime<120) {
                    student.run();
                    if (counter > 0) {
                        counter--;
                    }
                } else {
                    System.out.println(" 考试时间到，全部交卷！");
                    Student tmpStudent;
                    for (Iterator<Student> iterator2 = students.iterator(); iterator2.hasNext();) {
                        tmpStudent = iterator2.next();
                        tmpStudent.setForce(true);
                        tmpStudent.run();
                        if (counter > 0) {
                            counter--;
                        }
                    }
                }
            }
        } catch (Exception e) {
            // TODO: handle exception
            e.printStackTrace();
        }
    }
 
}