java - 如何实现带回调的异步调用？

Question

我需要创建一个库，其中包含同步和异步方法。

我的库的核心逻辑 -

客户将使用我们的库，他们将通过传递 DataKey 构建器对象来调用它。然后，我们将使用该 DataKey 对象构造一个 URL，并通过执行它来对该 URL 进行 HTTP 客户端调用，在我们以 JSON 字符串形式返回响应后，我们将该 JSON 字符串发送回我们的客户通过创建 DataResponse 对象来实现。

我将有同步和异步方法。有些客户将调用 executeSynchronous 方法来获得相同的功能，有些客户将调用我们的 executeAsynchronous 方法，并且使用 executeAsynchronous 方法，他们将调用future.get 在代码本身中。

下面是我的界面 -

public interface Client {

    // for synchronous
    public DataResponse executeSynchronous(DataKey dataKey);

    // for asynchronous
    public Future<DataResponse> executeAsynchronous(DataKey dataKey);
}

下面是我的 DataResponse 类 -

public class DataResponse {

    private String response;
    private DataErrorEnum error;
    private DataStatusEnum status;

    // constructor here

    // and getters here
}

下面是我的 DataStatusEnum 类 -

public enum DataStatusEnum {
    SUCCESS, ERROR;
}

下面是我的 DataErrorEnum 类 -

public enum DataErrorEnum {
    NONE(200, "NONE", "Response is success."),
    SERVER_DOWN(3145, "Server Down", "some long message here which can give more details."),
    CLIENT_ERROR(3123, "Client Error", "some long message here which can give more details."),
    TIMEOUT_ON_CLIENT(3187, "Client Timeout", "some long message here which can give more details.");

    private final int code;
    private final String status;
    private final String description;

    // constructor and getters here
}

然后我就有了实现上述 Client 接口(interface)的 DataClient。

public class DataClient implements Client {

    private RestTemplate restTemplate = new RestTemplate();
    private ExecutorService service = Executors.newFixedThreadPool(10);

    // for synchronous call
    @Override
    public DataResponse executeSynchronous(DataKey dataKey) {
        DataResponse dataResponse = null;

        try {
            Future<String> future = executeAsynchronous(dataKey);
            dataResponse = future.get(dataKey.getTimeout(), TimeUnit.MILLISECONDS);
        } catch (TimeoutException ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.TIMEOUT_ON_CLIENT, dataKey);
            dataResponse = new DataResponse(null, DataErrorEnum.TIMEOUT_ON_CLIENT, DataStatusEnum.ERROR);
        } catch (Exception ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.CLIENT_ERROR, dataKey);
            dataResponse = new DataResponse(null, DataErrorEnum.CLIENT_ERROR, DataStatusEnum.ERROR);
        }

        return dataResponse;
    }

    //for asynchronous call
    @Override
    public Future<DataResponse> executeAsynchronous(DataKey dataKey) {
        Future<DataResponse> future = null;

        try {
            Task task = new Task(dataKey, restTemplate);
            future = executor.submit(task);
        } catch (Exception ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.CLIENT_ERROR, dataKey);
        }

        return future;
    }
}

现在下面是我的简单类，它将执行实际任务 -

public class Task implements Callable<DataResponse> {

    private DataKey dataKey;
    private RestTemplate restTemplate;

    public Task(DataKey dataKey, RestTemplate restTemplate) {
        this.dataKey = dataKey;
        this.restTemplate = restTemplate;
    }

    @Override
    public DataResponse call() throws Exception {
        DataResponse dataResponse = null;
        String response = null;

        try {
            String url = createURL();
            response = restTemplate.getForObject(url, String.class);

            // it is a successful response
            dataResponse = new DataResponse(response, DataErrorEnum.NONE, DataStatusEnum.SUCCESS);
        } catch (RestClientException ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.SERVER_DOWN, dataKey);
            dataResponse = new DataResponse(null, DataErrorEnum.SERVER_DOWN, DataStatusEnum.ERROR);
        } catch (Exception ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.CLIENT_ERROR, dataKey);
            dataResponse = new DataResponse(null, DataErrorEnum.CLIENT_ERROR, DataStatusEnum.ERROR);
        }

        return dataResponse;
    }

    // create a URL by using dataKey object
    private String createURL() {
        String url = somecode;

        return url;
    }
}

问题陈述:-

正如我上面提到的，一些客户将调用 executeSynchronous 方法来获取他们在 DataKey 对象中传递的用户 ID 的数据，而一些客户将调用 >executeAsynchronous 方法与 DataKey 对象，但在后一种情况下，它们将在其代码库中执行 future.get 。

如果您看到我的 executeSynchronous 方法，则说明我正在调用 executeAsynchronous 方法后执行 future.get，并且是否存在任何 TimeoutException ，然后我使用我们公司特定的 PotoLogging 类进行日志记录，该日志将转到此处的其他一些服务，我们用它来查看仪表板上的所有错误日志。这主要取决于我们如何使用什么名称来记录它，以便我们可以在仪表板中看到这些名称。

现在的问题是我们公司内的客户也可以调用 executeAsynchronous 方法，但这意味着他们将在其代码库中执行 future.get ，这也可能导致TimeoutException 在他们的代码中，但我不能强制他们以与我相同的方式记录。所以我的问题是 - 如果有任何 TimeoutException ，我是否可以通过任何方式获得回调，以便如果有人调用我的库的 executeAsynchronous 方法，我可以像这样记录它在他们的代码库中 -

PotoLogging.logErrors(ex, DataErrorEnum.TIMEOUT_ON_CLIENT, dataKey);

我需要这样做，以便我的库可以按照我们想要的方式在我们公司拥有的工具中记录 TimeoutException。否则，我需要告诉每个客户像这样记录它，以便我们可以在仪表板中看到它。如何从异步调用中获取回调并仍然利用异步的所有功能？

最好的方法是什么？

Answer 1

Future 只是一个接口(interface)。提供一个包装服务返回的实例的实现。让它将所有调用委托(delegate)给实际的 Future 并用适当的 try-catch block 包装这些调用。

Future<DataResponse> wrapper = new Future<DataResponse>() {
    private final Future<DataResponse> delegate = future;

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return delegate.cancel(mayInterruptIfRunning);
    }

    @Override
    public boolean isCancelled() {
        return delegate.isCancelled();
    }

    @Override
    public boolean isDone() {
        return delegate.isDone();
    }

    @Override
    public DataResponse get() throws InterruptedException, ExecutionException {
        DataResponse dataResponse = null;
        try {
            delegate.get();
        } catch (TimeoutException ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.TIMEOUT_ON_CLIENT, dataKey);
            dataResponse = new DataResponse(null, DataErrorEnum.TIMEOUT_ON_CLIENT, DataStatusEnum.ERROR);
        }
        return dataResponse;
    }

    @Override
    public DataResponse get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        DataResponse dataResponse = null;
        try {
            delegate.get(timeout, unit);
        } catch (TimeoutException ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.TIMEOUT_ON_CLIENT, dataKey);
            dataResponse = new DataResponse(null, DataErrorEnum.TIMEOUT_ON_CLIENT, DataStatusEnum.ERROR);
        }
        return dataResponse;
    }
};
return wrapper;