java - 如何停止 future 的超时

Question

我正在计算等待串行事件发生超时的 future :

Future<Response> future = executor.submit(new CommunicationTask(this, request));
response = new Response("timeout");
try {
  response = future.get(timeoutMilliseconds, TimeUnit.MILLISECONDS);
} catch (InterruptedException | TimeoutException e) {
  future.cancel(true);
  log.info("Execution time out." + e);
} catch (ExecutionException e) {
  future.cancel(true);
  log.error("Encountered problem communicating with device: " + e);
}

CommunicationTask 类实现了Observer 接口(interface)来监听串口的变化。

问题是从串行端口读取相对较慢，即使发生串行事件，时间也会用完并抛出 TimeoutException。当连续事件发生时，我该怎么做才能停止我 future 的超时时钟？

我用 AtomicReference 尝试过，但没有任何改变:

public class CommunicationTask implements Callable<Response>, Observer {
  private AtomicReference atomicResponse = new AtomicReference(new Response("timeout"));
  private CountDownLatch latch = new CountDownLatch(1);
  private SerialPort port;

  CommunicationTask(SerialCommunicator communicator, Request request) {
    this.communicator = communicator;
    this.message = request.serialize();
    this.port = communicator.getPort();
  }

  @Override
  public Response call() throws Exception {
    return query(message);
  }

  public Response query(String message) {
    communicator.getListener().addObserver(this);
    message = message + "\r\n";
    try {
      port.writeString(message);
    } catch (Exception e) {
      log.warn("Could not write to port: " + e);
      communicator.disconnect();
    }
    try {
      latch.await();
    } catch (InterruptedException e) {
      log.info("Execution time out.");
    }
    communicator.getListener().deleteObserver(this);
    return (Response)atomicResponse.get();
  }

  @Override
  public void update(Observable o, Object arg) {
    atomicResponse.set((Response)arg);
    latch.countDown();
  }
}

我该怎么做才能解决这个问题？

编辑:

好的，我有一个错误。在我的 update 函数中设置 atomicResponse 之前，我正在倒计时。现在它似乎可行，但仍然存在这样的问题，即这种方法是否是正确的方法？

Answer 1

你有没有探索过谷歌的 Guava 'future listener'，它是基于 Async future 的，希望下面的代码片段能帮助你....

import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

public class SyncFutureExample {
    public static void main(String[] args) {
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1));
        ListenableFuture<String> lf  = service.submit(new CommuncationTask());

        //no need for future.get() or future.get(10,time minutes)


        //add callbacks(= async future listeners) ....
        Futures.addCallback(lf, new FutureCallback<String>() {
              public void onSuccess(String input) {
                System.out.println(input + " >>> success");//gets a callback once task is success
              }
              public void onFailure(Throwable thrown) {
                  System.out.println(thrown + " >>> failure");//gets a callback if task is failed
              }
            });
        service.shutdown();
    }

}

class CommuncationTask implements Callable<String>{

    public String call() throws Exception {
        TimeUnit.SECONDS.sleep(15);// some dummy serious task .............
        return "TaskDone";
    }


}