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<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();
    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