最大重试次数后异步 HTTP 发布重试和耗尽

Question

如何在多次失败的异步请求调用后耗尽重试次数？ 我正在使用AsyncHttpClient 向我们的服务器发送请求。如果出现请求超时、连接异常等情况，我希望客户端重试 N 次并抛出自定义异常。调用方法应该收到此异常，否则可以不处理。

// calls post
public void call(String data) throws CustomException {
    
    asyncHttpClient.post(data, 10);

}

// posts data to http endpoint
public void post(String data, int retries) throw CustomException {
    // if retries are exhausted, throw CustomException to call()
    if (retry <= 0) {
        throw new CustomException("exc");
    }

    BoundRequest request = httpClient.preparePost("http_endpoint");
    ListenableFuture<Response> responseFuture = httpClient.post(request);
 
    responseFuture.addListener(() -> {
        Response response = null;
        int status = 0;
        try {

            response = responseFuture.get();
            status = response.getStatusCode();

            // HTTP_ACCEPTED = {200, 201, 202}
            if (ArrayUtils.contains(HTTP_ACCEPTED, status)) {
                // ok
            } else {
                sleep(10);
                post(data, retry - 1);
            }
        } catch (InterruptedException e) {
            sleep(10);
            post(data, retry - 1);

        } catch (ExecutionException e) {
            // ConnectionException
            // RequestTimeoutException
            sleep(10); // 10 seconds
            post(data, retry - 1);
        } catch (Exception e) {
            sleep(10); // 10 seconds
            post(data, retry - 1 );
        } finally {
            responseFuture.done();
        }
    },  Runnable::run);

} 

这种方法有几个问题：

1. 使用递归调用重试。
2. CustomException 似乎从未被抛出，并且在重试 == 0 后，控件返回到 finally 块。

...
} catch (ExecutionException e) {
    // ConnectionException
    // RequestTimeoutException
    sleep(10); // 10 seconds
    try {
        post(data, retry - 1);
    } catch (CustomException e) {
    }
}
...

Answer 1

AsyncHttpClient中有一个预定义的函数来处理MaxRetries，

下面的代码展示了一个简单的实现

AsyncHttpClientConfig cf = new DefaultAsyncHttpClientConfig.Builder().setMaxRequestRetry(5).setKeepAlive(true).build()
final AsyncHttpClient asyncHttpClient = new DefaultAsyncHttpClient(cf);

你可以去掉你的重试逻辑，让AsyncHttpClient来处理。

Answer 2

好的，所以尝试重现您试图用您的代码实现的目标，但立即意识到您的CustomException 仅适用于RuntimeException 类型。原因是你想在运行时和另一个线程中抛出异常。

下面的代码显示了异常的简单实现。请记住，并非所有 RuntimeExceptions 都会停止程序。这在thread 中有解释。所以如果你想终止程序，你必须手动停止它。

public class CustomException extends RuntimeException {

    public CustomException(String msg) {
        super(msg);
        // print your exception to the console
 
        // optional: exit the program
        System.exit(0);
    }

}

我更改了您实现的其余部分，这样您就不必再进行递归调用了。我删除了回调方法，而是调用了get() 方法，该方法等待请求完成。但由于我在一个单独的线程中执行所有这些，它应该在后台而不是主线程中运行。

public class Main {

    private final AsyncHttpClient httpClient;
    private final int[] HTTP_ACCEPTED = new int[]{200, 201, 202};

    private final static String ENDPOINT = "https://postman-echo.com/post";

    public static void main(String[] args) {
        String data = "{message: 'Hello World'}";
        Main m = new Main();
        m.post(data, 10);

    }

    public Main() {
        httpClient = asyncHttpClient();
    }

    public void post(final String data, final int retry) {

        Runnable runnable = () -> {
            int retries = retry;

            for (int i = 0; i < retry; i++) {
                Request request = httpClient.preparePost(ENDPOINT)
                        .addHeader("Content-Type", "application/json")
                        .setBody(data)
                        .build();

                ListenableFuture<Response> responseFuture = httpClient.executeRequest(request);
                try {
                    Response response = responseFuture.get();
                    int status = response.getStatusCode();

                    if (ArrayUtils.contains(HTTP_ACCEPTED, status)) {
                        System.out.println("Successful! Breaking Loop");
                        break;
                    } else {
                        Thread.sleep(10);
                    }

                } catch (InterruptedException | ExecutionException ex) {
                    Logger.getLogger(Main.class.getName()).log(Level.SEVERE, null, ex);
                }
                retries--;
            }

            System.out.println("Remaining retries: " + retries);

            if (retries <= 0) {
                throw new CustomException("exc");
            }
        };

        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.submit(runnable);
    }

}

另类

您可以使用相同的Runnable 进行异步调用，而无需等待future.get()。每个侦听器都可以方便地在同一个线程中调用，这对于您的用例来说更加高效。

public void post2(final String data, final int retry) {
        Request request = httpClient.preparePost(ENDPOINT)
                .addHeader("Content-Type", "application/json")
                .setBody(data)
                .build();

        ListenableFuture<Response> future = httpClient.executeRequest(request);

        MyRunnable runnable = new MyRunnable(retry, future, request);
        future.addListener(runnable, null);
}

public class MyRunnable implements Runnable {

        private int retries;
        private ListenableFuture<Response> responseFuture;
        private final Request request;

        public MyRunnable(int retries, ListenableFuture<Response> future, Request request) {
            this.retries = retries;
            this.responseFuture = future;
            this.request = request;
        }

        @Override
        public void run() {

            System.out.println("Remaining retries: " + this.retries);
            System.out.println("Thread ID: " + Thread.currentThread().getId());


            try {
                Response response = responseFuture.get();
                int status = response.getStatusCode();

                if (ArrayUtils.contains(HTTP_ACCEPTED, status)) {
                    System.out.println("Success!");
                    //do something here

                } else if (this.retries > 0) {
                    Thread.sleep(10);
                    this.execute();
                } else {
                    throw new CustomException("Exception!");
                }

            } catch (InterruptedException | ExecutionException e) {
                this.execute();
            }
        }

        private void execute() {
            this.retries -= 1;
            this.responseFuture = httpClient.executeRequest(this.request);
            this.responseFuture.addListener(this, null);
        }
}

Answer 3

在某些情况下，很多人不想 throw e 而不是 throw new RuntimeException (ee)

catch (ExecutionException e) {
    Throwable ee = e.getCause ();

    if (ee instanceof InvalidInputException)
    {
        //error handling 1
    } else if (ee instanceof MiscalculationException e)
    {
        //error handling 2
    }
    else throw e; // Not ee here
}