Java中進(jìn)行異步調(diào)用失敗的解決方法詳解

更新時(shí)間：2026年01月13日 09:23:01 作者：程序員大華

異步編程是一種非阻塞的編程模式,允許程序在等待某個(gè)操作完成時(shí)繼續(xù)執(zhí)行其他任務(wù),而不是一直等待,下面我們就來(lái)看看異步調(diào)用失敗的相關(guān)解決方法吧

1.異步編程介紹

什么是異步編程

異步編程是一種非阻塞的編程模式，允許程序在等待某個(gè)操作完成時(shí)繼續(xù)執(zhí)行其他任務(wù)，而不是一直等待。

當(dāng)操作完成后，通過(guò)回調(diào)函數(shù)、Future 或事件通知等方式獲取結(jié)果。

同步 vs 異步對(duì)比：

同步：順序執(zhí)行，每一步必須等待前一步完成
異步：非阻塞執(zhí)行，可以同時(shí)處理多個(gè)任務(wù)

Java 中的異步實(shí)現(xiàn)方式

CompletableFuture (Java 8+)

// 創(chuàng)建異步任務(wù)
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 模擬耗時(shí)操作
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
    return "異步任務(wù)結(jié)果";
});

// 處理結(jié)果
future.thenAccept(result -> System.out.println("結(jié)果: " + result));

@Async 注解 (Spring Framework)

@Service
public class AsyncService {
    
    @Async
    public CompletableFuture<String> asyncMethod() {
        // 異步執(zhí)行的方法
        return CompletableFuture.completedFuture("執(zhí)行結(jié)果");
    }
}

回調(diào)函數(shù)

public interface Callback {
    void onSuccess(String result);
    void onError(Exception e);
}

public void asyncOperation(Callback callback) {
    new Thread(() -> {
        try {
            // 模擬操作
            String result = doSomething();
            callback.onSuccess(result);
        } catch (Exception e) {
            callback.onError(e);
        }
    }).start();
}

2.異步編程中的常見(jiàn)錯(cuò)誤

網(wǎng)絡(luò)相關(guān)錯(cuò)誤

連接超時(shí)
讀取超時(shí)
DNS 解析失敗
網(wǎng)絡(luò)不可達(dá)

資源相關(guān)錯(cuò)誤

內(nèi)存不足
線(xiàn)程池耗盡
數(shù)據(jù)庫(kù)連接超時(shí)

業(yè)務(wù)邏輯錯(cuò)誤

遠(yuǎn)程服務(wù)返回錯(cuò)誤碼
數(shù)據(jù)格式異常
業(yè)務(wù)規(guī)則校驗(yàn)失敗

示例：可能出錯(cuò)的異步方法

public class UnreliableService {
    
    // 模擬不可靠的遠(yuǎn)程服務(wù)調(diào)用
    public CompletableFuture<String> callExternalService(String data) {
        return CompletableFuture.supplyAsync(() -> {
            // 模擬隨機(jī)錯(cuò)誤
            double random = Math.random();
            if (random < 0.3) {
                throw new RuntimeException("網(wǎng)絡(luò)超時(shí)");
            } else if (random < 0.6) {
                throw new RuntimeException("服務(wù)端錯(cuò)誤");
            }
            return "處理結(jié)果: " + data;
        });
    }
}

3. 異步重試機(jī)制實(shí)現(xiàn)

手動(dòng)重試實(shí)現(xiàn)

基礎(chǔ)重試邏輯

public class SimpleRetry {
    
    public static <T> CompletableFuture<T> retryAsync(
            Supplier<CompletableFuture<T>> task, 
            int maxAttempts, 
            long delayMs) {
        
        CompletableFuture<T> result = new CompletableFuture<>();
        retryAsync(task, maxAttempts, delayMs, 1, result);
        return result;
    }
    
    private static <T> void retryAsync(
            Supplier<CompletableFuture<T>> task, 
            int maxAttempts, 
            long delayMs, 
            int attempt, 
            CompletableFuture<T> result) {
        
        task.get().whenComplete((response, throwable) -> {
            if (throwable == null) {
                result.complete(response);
            } else if (attempt >= maxAttempts) {
                result.completeExceptionally(throwable);
            } else {
                // 延遲后重試
                CompletableFuture.delayedExecutor(delayMs, TimeUnit.MILLISECONDS)
                    .execute(() -> retryAsync(task, maxAttempts, delayMs, attempt + 1, result));
            }
        });
    }
}

使用 Spring Retry

添加依賴(lài)

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
    <version>2.0.0</version>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-aspects</artifactId>
    <version>5.3.0</version>
</dependency>

配置重試模板

@Configuration
@EnableRetry
public class RetryConfig {
    
    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate template = new RetryTemplate();
        
        // 重試策略：最多重試3次，遇到特定異常時(shí)重試
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(3, 
            Collections.singletonMap(RuntimeException.class, true));
        template.setRetryPolicy(retryPolicy);
        
        // 退避策略：每次重試間隔1秒
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1000);
        template.setBackOffPolicy(backOffPolicy);
        
        return template;
    }
}

使用 @Retryable 注解

@Service
public class RetryableService {
    
    @Retryable(
        value = {RuntimeException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000)
    )
    @Async
    public CompletableFuture<String> retryableAsyncMethod() {
        return CompletableFuture.supplyAsync(() -> {
            // 模擬可能失敗的操作
            if (Math.random() < 0.7) {
                throw new RuntimeException("臨時(shí)錯(cuò)誤");
            }
            return "成功結(jié)果";
        });
    }
    
    @Recover
    public CompletableFuture<String> recover(RuntimeException e) {
        return CompletableFuture.completedFuture("降級(jí)結(jié)果");
    }
}

高級(jí)重試策略實(shí)現(xiàn)

指數(shù)退避重試

public class ExponentialBackoffRetry {
    
    public static <T> CompletableFuture<T> retryWithExponentialBackoff(
            Supplier<CompletableFuture<T>> task,
            int maxAttempts,
            long initialDelay,
            long maxDelay) {
        
        return retryWithExponentialBackoff(task, maxAttempts, initialDelay, maxDelay, 1);
    }
    
    private static <T> CompletableFuture<T> retryWithExponentialBackoff(
            Supplier<CompletableFuture<T>> task,
            int maxAttempts,
            long initialDelay,
            long maxDelay,
            int attempt) {
        
        CompletableFuture<T> future = task.get();
        
        if (attempt >= maxAttempts) {
            return future;
        }
        
        return future.handle((result, throwable) -> {
            if (throwable == null) {
                return CompletableFuture.completedFuture(result);
            } else {
                // 計(jì)算退避時(shí)間
                long delay = Math.min(maxDelay, initialDelay * (long) Math.pow(2, attempt - 1));
                
                // 添加隨機(jī)抖動(dòng)避免驚群效應(yīng)
                delay = (long) (delay * (0.5 + Math.random()));
                
                CompletableFuture<T> nextAttempt = CompletableFuture
                    .delayedExecutor(delay, TimeUnit.MILLISECONDS)
                    .supplyAsync(() -> 
                        retryWithExponentialBackoff(task, maxAttempts, initialDelay, maxDelay, attempt + 1)
                    )
                    .thenCompose(cf -> cf);
                
                return nextAttempt;
            }
        }).thenCompose(cf -> cf);
    }
}

基于條件的重試

public class ConditionalRetry {
    
    public static <T> CompletableFuture<T> retryWithCondition(
            Supplier<CompletableFuture<T>> task,
            Predicate<Throwable> shouldRetry,
            int maxAttempts,
            Function<Integer, Long> delayCalculator) {
        
        CompletableFuture<T> result = new CompletableFuture<>();
        retryWithCondition(task, shouldRetry, maxAttempts, delayCalculator, 1, result);
        return result;
    }
    
    private static <T> void retryWithCondition(
            Supplier<CompletableFuture<T>> task,
            Predicate<Throwable> shouldRetry,
            int maxAttempts,
            Function<Integer, Long> delayCalculator,
            int attempt,
            CompletableFuture<T> result) {
        
        task.get().whenComplete((response, throwable) -> {
            if (throwable == null) {
                result.complete(response);
                return;
            }
            
            boolean canRetry = attempt < maxAttempts && shouldRetry.test(throwable);
            
            if (!canRetry) {
                result.completeExceptionally(throwable);
                return;
            }
            
            long delay = delayCalculator.apply(attempt);
            CompletableFuture.delayedExecutor(delay, TimeUnit.MILLISECONDS)
                .execute(() -> 
                    retryWithCondition(task, shouldRetry, maxAttempts, delayCalculator, attempt + 1, result)
                );
        });
    }
}

完整的重試工具類(lèi)

public class AsyncRetryUtil {
    
    /**
     * 異步重試執(zhí)行器
     */
    public static class AsyncRetryBuilder<T> {
        private final Supplier<CompletableFuture<T>> task;
        private int maxAttempts = 3;
        private Predicate<Throwable> retryCondition = ex -> true;
        private Function<Integer, Long> delayStrategy = attempt -> 1000L * attempt;
        private Consumer<Integer> onRetry = attempt -> {};
        private Function<Throwable, T> fallback = null;
        
        private AsyncRetryBuilder(Supplier<CompletableFuture<T>> task) {
            this.task = task;
        }
        
        public static <T> AsyncRetryBuilder<T> of(Supplier<CompletableFuture<T>> task) {
            return new AsyncRetryBuilder<>(task);
        }
        
        public AsyncRetryBuilder<T> maxAttempts(int maxAttempts) {
            this.maxAttempts = maxAttempts;
            return this;
        }
        
        public AsyncRetryBuilder<T> retryIf(Predicate<Throwable> condition) {
            this.retryCondition = condition;
            return this;
        }
        
        public AsyncRetryBuilder<T> delayStrategy(Function<Integer, Long> strategy) {
            this.delayStrategy = strategy;
            return this;
        }
        
        public AsyncRetryBuilder<T> onRetry(Consumer<Integer> callback) {
            this.onRetry = callback;
            return this;
        }
        
        public AsyncRetryBuilder<T> fallback(Function<Throwable, T> fallback) {
            this.fallback = fallback;
            return this;
        }
        
        public CompletableFuture<T> execute() {
            CompletableFuture<T> result = new CompletableFuture<>();
            execute(1, result);
            return result;
        }
        
        private void execute(int attempt, CompletableFuture<T> result) {
            task.get().whenComplete((response, throwable) -> {
                if (throwable == null) {
                    result.complete(response);
                    return;
                }
                
                boolean shouldRetry = attempt < maxAttempts && retryCondition.test(throwable);
                
                if (!shouldRetry) {
                    if (fallback != null) {
                        try {
                            T fallbackResult = fallback.apply(throwable);
                            result.complete(fallbackResult);
                        } catch (Exception e) {
                            result.completeExceptionally(e);
                        }
                    } else {
                        result.completeExceptionally(throwable);
                    }
                    return;
                }
                
                onRetry.accept(attempt);
                long delay = delayStrategy.apply(attempt);
                
                CompletableFuture.delayedExecutor(delay, TimeUnit.MILLISECONDS)
                    .execute(() -> execute(attempt + 1, result));
            });
        }
    }
}

使用示例

public class RetryExample {
    
    public static void main(String[] args) throws Exception {
        // 模擬不可靠的服務(wù)
        Supplier<CompletableFuture<String>> unreliableService = () -> 
            CompletableFuture.supplyAsync(() -> {
                double rand = Math.random();
                if (rand < 0.8) {
                    throw new RuntimeException("服務(wù)暫時(shí)不可用");
                }
                return "服務(wù)調(diào)用成功";
            });
        
        // 使用重試工具
        CompletableFuture<String> result = AsyncRetryUtil.AsyncRetryBuilder
            .of(unreliableService)
            .maxAttempts(5)
            .retryIf(ex -> ex.getMessage().contains("暫時(shí)不可用"))
            .delayStrategy(attempt -> 1000L * attempt) // 線(xiàn)性退避
            .onRetry(attempt -> 
                System.out.println("第 " + attempt + " 次重試..."))
            .fallback(ex -> "降級(jí)結(jié)果")
            .execute();
        
        result.whenComplete((response, error) -> {
            if (error != null) {
                System.out.println("最終失敗: " + error.getMessage());
            } else {
                System.out.println("最終結(jié)果: " + response);
            }
        });
        
        // 等待異步操作完成
        Thread.sleep(10000);
    }
}

4. 注意事項(xiàng)

重試策略選擇

網(wǎng)絡(luò)超時(shí)：使用指數(shù)退避 + 隨機(jī)抖動(dòng)
服務(wù)限流：根據(jù)返回的等待時(shí)間重試
業(yè)務(wù)錯(cuò)誤：根據(jù)具體錯(cuò)誤碼決定是否重試

避免的問(wèn)題

無(wú)限重試：設(shè)置最大重試次數(shù)
資源耗盡：合理控制重試頻率
雪崩效應(yīng)：使用斷路器模式配合重試
重復(fù)操作：確保操作的冪等性

監(jiān)控和日志

// 添加重試監(jiān)控
public class RetryMonitor {
    private static final MeterRegistry meterRegistry = new SimpleMeterRegistry();
    
    public static void recordRetry(String operation, int attempt) {
        Counter.builder("retry.attempts")
            .tag("operation", operation)
            .register(meterRegistry)
            .increment();
    }
    
    public static void recordSuccess(String operation, long duration) {
        Timer.builder("retry.duration")
            .tag("operation", operation)
            .tag("status", "success")
            .register(meterRegistry)
            .record(duration, TimeUnit.MILLISECONDS);
    }
}

具體業(yè)務(wù)場(chǎng)景選擇合適的重試策略，提高系統(tǒng)的容錯(cuò)能力和穩(wěn)定性。

以上就是Java中進(jìn)行異步調(diào)用失敗的解決方法詳解的詳細(xì)內(nèi)容，更多關(guān)于Java異步調(diào)用的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: