如何将异步HTTP请求抽象为同步

Question

我有一个第三方REST服务，它的行为是“异步的”；就像在请求中使用中间响应进行响应一样，中间响应由与中间响应相关的回调补充。

回调几乎立即通过“回调URL”返回。

我试图设计一种解决方案，某种程度上是一种适配器来调用这个资源，就好像它是“同步的”，因为处理回调非常麻烦，特别是当我需要通过同一个第三方顺序处理其他类似API的连续请求时。基本上，我希望抽象绿色部分，以便调用方只获得完全回调、错误或超时异常。

​

我的研究重点是使用RxJava，但我不知道如何用反应性编程的原则来解决这个问题(我的理解是有限的)。

设计考虑因素：

1. 将相关器持久化到数据存储以供以后在回调时查找是不可取的，因为它很昂贵。
2. 回调的等待策略是可以的，因为回调的响应时间小于1秒。

如何使用CompletableFuture或Observable-Observer模式等待回调并返回给调用方？

Answer 1

考虑使用CountDownLatch对主线程进行等待，直到工作线程处理了第三方API。工作人员获得回调，以便知道请求何时已取得进展、是否完成、超时、是否有错误等。

下面是一个粗略的模拟：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;

public class TreatAsSync {

    public static void main(String[] args) {
        TreatAsSync app = new TreatAsSync();
        app.call();
    }

    private void call() {
        RestClient restClient = new RestClient();
        Request request = new Request();
        Response response = restClient.call(request);
        System.out.println("Response was: " + response);
    }

    private class Request {

    }

    private class Response {
        private final boolean error;
        private final boolean timedOut;
        private final String result;

        public Response(boolean error, boolean timedOut, String result) {
            this.error = error;
            this.timedOut = timedOut;
            this.result = result;
        }

        public String toString() {
            return "error:" + error + ", timedOut: " + timedOut + ", result: " + result;
        }
    }

    private class ResponseWrapper {
        private Response response;

        public Response getResponse() {
            return response;
        }

        public void setRespose(Response response) {
            this.response = response;
        }
    }

    private class RestClient {

        public Response call(Request request) {
            ResponseWrapper wrapper = new ResponseWrapper();
            CountDownLatch latch = new CountDownLatch(1);

            ThirdPartyRunner runner = new ThirdPartyRunner(request, wrapper, latch);
            new Thread(runner).start();

            try {
                latch.await();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            return wrapper.getResponse();
        }
    }

    private interface ThirdPartyCallBack {
        public void onProgress(Response response);
        public void onComplete(Response response);
        public void onTimeOut(Response response);
        public void onError(Response response);
    }

    private class ThirdPartyRunner implements ThirdPartyCallBack, Runnable {
        private final Request request;
        private final ResponseWrapper wrapper;
        private final CountDownLatch latch;

        public ThirdPartyRunner(Request request, ResponseWrapper wrapper, CountDownLatch latch) {
            this.request = request;
            this.wrapper = wrapper;
            this.latch = latch;
        }

        @Override
        public void onProgress(Response response) {
            System.out.println("some progress was made...");
        }

        @Override
        public void onComplete(Response response) {
            System.out.println("request completed");
            finished(response);
        }

        @Override
        public void onTimeOut(Response response) {
            System.out.println("request timed out");
            finished(response);
        }

        @Override
        public void onError(Response response) {
            System.out.println("request had an error");
            finished(response);
        }

        private void finished(Response response) {
            wrapper.setRespose(response);
            latch.countDown();
        }

        @Override
        public void run() {
            try {
                callThirdParty();
            } catch (Exception e) {
                finished(new Response(true, false, e.getMessage()));
            }
        }

        private void callThirdParty() {
            // simulate api.call(request, this);
            for (int i = 0; i < ThreadLocalRandom.current().nextInt(10) + 1; i++) {
                onProgress(new Response(false, false, "in progress"));
            }

            switch (ThreadLocalRandom.current().nextInt(3)) {
            case 0:
                onComplete(new Response(false, false, "done"));
                break;

            case 1:
                onTimeOut(new Response(false, true, "hello?"));
                break;

            case 2:
                onError(new Response(true, false, "uh oh!"));
                break;
            }
        }
    }

}