Java超时控制的实现

摘要：基本原理采用的和方法在另外一个线程中结果回来，一下，返回否则就等待超时返回超时采用一线程轮询的的双重保险实例参考

采用LockSupport的parkNanos和unpack方法

在另外一个线程中结果回来，unpack一下，返回；否则就等待超时返回（超时采用一线程轮询 + lock的condition的await 双重保险）

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
 * 
 * Created by codecraft on 2015/8/26.
 */
public class DefaultFuture {
    private static final Map FUTURES = new ConcurrentHashMap();
    private final long id;
    private final Lock lock = new ReentrantLock();
    private final Condition done = lock.newCondition();
    private volatile Response response;
    private final long start = System.currentTimeMillis();
    private final int timeout;
    public DefaultFuture(long id,int timeout) {
        this.id = id;
        this.timeout = timeout;
    }
    private long getStartTimestamp() {
        return start;
    }
    public int getTimeout() {
        return timeout;
    }
    public boolean isDone() {
        return response != null;
    }
    public long getId() {
        return id;
    }
    public Object get(int timeout){
        if (timeout <= 0) {
            timeout = 1000;
        }
        if (! isDone()) {
            long start = System.currentTimeMillis();
            lock.lock();
            try {
                while (! isDone()) {
                    done.await(timeout, TimeUnit.MILLISECONDS);
                    if (isDone() || System.currentTimeMillis() - start > timeout) {
                        break;
                    }
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }
            if (! isDone()) {
//                throw new RuntimeException("timeout");
                System.out.println("timeout");
            }
        }
        return response;
    }
    private void doReceived(Response res) {
        lock.lock();
        try {
            response = res;
            if (done != null) {
                done.signal();
            }
        } finally {
            lock.unlock();
        }
    }
    public static void received(Response response) {
        try {
            DefaultFuture future = FUTURES.remove(response.getId());
            if (future != null) {
                future.doReceived(response);
            } else {
                System.out.println("The timeout response finally returned at "
                        + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
                        + ", response ");
            }
        } finally {
//            CHANNELS.remove(response.getId());
        }
    }
    private static class RemotingInvocationTimeoutScan implements Runnable {
        public void run() {
            while (true) {
                try {
                    for (DefaultFuture future : FUTURES.values()) {
                        if (future == null || future.isDone()) {
                            continue;
                        }
                        if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) {
                            // create exception response.
                            Response timeoutResponse = new Response(future.getId());
                            // handle response.
                            DefaultFuture.received(timeoutResponse);
                        }
                    }
                    Thread.sleep(30);
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }
        }
    }
    static {
        Thread th = new Thread(new RemotingInvocationTimeoutScan(), "ResponseTimeoutScanTimer");
        th.setDaemon(true);
        th.start();
    }
    public static void main(String[] args){
        int timeout = 1000;
        System.out.println("start");
        final long start = System.currentTimeMillis();
        final DefaultFuture future = new DefaultFuture(1,timeout);
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (System.currentTimeMillis() - start < 2000) {
                    //sleep
                }
                Response response = new Response();
                response.setResult("hello");
                future.doReceived(response);
            }
        }).start();
        Object response = future.get(timeout);
        System.out.println(System.currentTimeMillis() - start);
        System.out.println("res "+response);
    }
}

dubbo-DefaultFuture