java concurrent 之 DelayQueue

DelayQueue

Delayed 元素的一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部 是延迟期满后保存时间最长的 Delayed 元素。如果延迟都还没有期满，则队列没有头部，并且 poll 将返回 null。当一个元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于等于 0 的值时，将发生到期。即使无法使用 take 或 poll 移除未到期的元素，也不会将这些元素作为正常元素对待。例如，size 方法同时返回到期和未到期元素的计数。此队列不允许使用 null 元素。

延迟队列的用途主要有以下：

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

现在实现一个简单的缓存失效的一个demo

cache

package com.viashare.delayedqueue;

import sun.jvm.hotspot.jdi.PrimitiveValueImpl;

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

/**
 * Created by Jeffy on 15/12/11.
 */
public class Cache<K, E> {

    private ConcurrentHashMap<K, E> concurrentHashMap = new ConcurrentHashMap<K, E>();


    private DelayQueue<Iterm<Key>> delayQueue = new DelayQueue<>();

    public Cache() {
        Thread checkThread =  timeoutCheck();
        checkThread.setDaemon(true);
        checkThread.setName("cache daemon thread");
        checkThread.start();
        checkThread.setUncaughtExceptionHandler((t,e)->{
            System.err.println(t.getName()+" had died ");
            timeoutCheck().start();
            e.printStackTrace();
        });
    }

    private Thread timeoutCheck() {
        return new Thread(() -> {
            System.err.println("A thread has started");
            while (true) {
                try {
                    Iterm<Key> iterm =  delayQueue.take();
                    System.err.println("delayed queue has remove the timeout element [ " + iterm.getKey()+" ]");
                    concurrentHashMap.remove(iterm.getKey());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    public E get(K key) {
        return concurrentHashMap.get(key);
    }


    public E put(K key, E value, long timeOut, TimeUnit timeUnit) {
        if (null == value) throw new IllegalArgumentException("value can not be null");

        delayQueue.put(new Iterm(key,timeOut));
        concurrentHashMap.put(key,value);
        return value;
    }


    static class Iterm<Key> implements Delayed {

        private static final long ORG_TIME = System.currentTimeMillis();

        private long triggerTime;

        private Key key;

        private long now() {
            return System.currentTimeMillis();
        }

        public Iterm(Key key, long timeout) {
            this.triggerTime = ORG_TIME + timeout;
            this.key = key;
        }

        public Key getKey() {
            return this.key;
        }

        public long getDelay(TimeUnit unit) {
            long d = unit.convert(triggerTime - now(), TimeUnit.MILLISECONDS);
            return d;
        }

        public int compareTo(Delayed o) {
            return 0;
        }
    }

    static class Key<K> {
        private K key;

        public Key(K key) {
            this.key = key;
        }
    }


}

测试类

public class CacheMain {

    public static void main(String[] args) throws InterruptedException {
        Cache cache = new Cache();
        cache.put("name","name",3000, TimeUnit.MILLISECONDS);
        Thread.sleep(1000);
        System.out.println(cache.get("name"));

        Thread.sleep(5000);
        System.out.println(cache.get("name"));

        System.err.println("finished........");

    }
}