1.异步编程介绍
什么是异步编程
异步编程是一种非阻塞的编程模式，允许程序在等待某个操作完成时继续执行其他任务，而不是一直等待。
当操作完成后，通过回调函数、Future 或事件通知等方式获取结果。
同步 vs 异步对比：
- 同步：顺序执行，每一步必须等待前一步完成
- 异步：非阻塞执行，可以同时处理多个任务
Java 中的异步实现方式
CompletableFuture (Java 8+)
// 创建异步任务
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
// 模拟耗时操作
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "异步任务结果";
});
// 处理结果
future.thenAccept(result -> System.out.println("结果: " + result));
@Async 注解 (Spring Framework)
@Service
public class AsyncService {
@Async
public CompletableFuture asyncMethod() {
// 异步执行的方法
return CompletableFuture.completedFuture("执行结果");
}
}
回调函数
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.异步编程中的常见错误
网络相关错误
- 连接超时
- 读取超时
- DNS 解析失败
- 网络不可达
资源相关错误
- 内存不足
- 线程池耗尽
- 数据库连接超时
业务逻辑错误
- 远程服务返回错误码
- 数据格式异常
- 业务规则校验失败
示例：可能出错的异步方法
public class UnreliableService {
// 模拟不可靠的远程服务调用
public CompletableFuture callExternalService(String data) {
return CompletableFuture.supplyAsync(() -> {
// 模拟随机错误
double random = Math.random();
if (random
3. 异步重试机制实现
手动重试实现
基础重试逻辑
public class SimpleRetry {
public static CompletableFuture retryAsync(
Supplier> task,
int maxAttempts,
long delayMs) {
CompletableFuture result = new CompletableFuture();
retryAsync(task, maxAttempts, delayMs, 1, result);
return result;
}
private static void retryAsync(
Supplier> task,
int maxAttempts,
long delayMs,
int attempt,
CompletableFuture 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
添加依赖
org.springframework.retryspring-retry2.0.0org.springframeworkspring-aspects5.3.0
配置重试模板
@Configuration
@EnableRetry
public class RetryConfig {
@Bean
public RetryTemplate retryTemplate() {
RetryTemplate template = new RetryTemplate();
// 重试策略：最多重试3次，遇到特定异常时重试
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 retryableAsyncMethod() {
return CompletableFuture.supplyAsync(() -> {
// 模拟可能失败的操作
if (Math.random() recover(RuntimeException e) {
return CompletableFuture.completedFuture("降级结果");
}
}
高级重试策略实现
指数退避重试
public class ExponentialBackoffRetry {
public static CompletableFuture retryWithExponentialBackoff(
Supplier> task,
int maxAttempts,
long initialDelay,
long maxDelay) {
return retryWithExponentialBackoff(task, maxAttempts, initialDelay, maxDelay, 1);
}
private static CompletableFuture retryWithExponentialBackoff(
Supplier> task,
int maxAttempts,
long initialDelay,
long maxDelay,
int attempt) {
CompletableFuture future = task.get();
if (attempt >= maxAttempts) {
return future;
}
return future.handle((result, throwable) -> {
if (throwable == null) {
return CompletableFuture.completedFuture(result);
} else {
// 计算退避时间
long delay = Math.min(maxDelay, initialDelay * (long) Math.pow(2, attempt - 1));
// 添加随机抖动避免惊群效应
delay = (long) (delay * (0.5 + Math.random()));
CompletableFuture 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 CompletableFuture retryWithCondition(
Supplier> task,
Predicate shouldRetry,
int maxAttempts,
Function delayCalculator) {
CompletableFuture result = new CompletableFuture();
retryWithCondition(task, shouldRetry, maxAttempts, delayCalculator, 1, result);
return result;
}
private static void retryWithCondition(
Supplier> task,
Predicate shouldRetry,
int maxAttempts,
Function delayCalculator,
int attempt,
CompletableFuture result) {
task.get().whenComplete((response, throwable) -> {
if (throwable == null) {
result.complete(response);
return;
}
boolean canRetry = attempt
retryWithCondition(task, shouldRetry, maxAttempts, delayCalculator, attempt + 1, result)
);
});
}
}
完整的重试工具类
public class AsyncRetryUtil {
/**
* 异步重试执行器
*/
public static class AsyncRetryBuilder {
private final Supplier> task;
private int maxAttempts = 3;
private Predicate retryCondition = ex -> true;
private Function delayStrategy = attempt -> 1000L * attempt;
private Consumer onRetry = attempt -> {};
private Function fallback = null;
private AsyncRetryBuilder(Supplier> task) {
this.task = task;
}
public static AsyncRetryBuilder of(Supplier> task) {
return new AsyncRetryBuilder(task);
}
public AsyncRetryBuilder maxAttempts(int maxAttempts) {
this.maxAttempts = maxAttempts;
return this;
}
public AsyncRetryBuilder retryIf(Predicate condition) {
this.retryCondition = condition;
return this;
}
public AsyncRetryBuilder delayStrategy(Function strategy) {
this.delayStrategy = strategy;
return this;
}
public AsyncRetryBuilder onRetry(Consumer callback) {
this.onRetry = callback;
return this;
}
public AsyncRetryBuilder fallback(Function fallback) {
this.fallback = fallback;
return this;
}
public CompletableFuture execute() {
CompletableFuture result = new CompletableFuture();
execute(1, result);
return result;
}
private void execute(int attempt, CompletableFuture result) {
task.get().whenComplete((response, throwable) -> {
if (throwable == null) {
result.complete(response);
return;
}
boolean shouldRetry = attempt execute(attempt + 1, result));
});
}
}
}
使用示例
public class RetryExample {
public static void main(String[] args) throws Exception {
// 模拟不可靠的服务
Supplier> unreliableService = () ->
CompletableFuture.supplyAsync(() -> {
double rand = Math.random();
if (rand result = AsyncRetryUtil.AsyncRetryBuilder
.of(unreliableService)
.maxAttempts(5)
.retryIf(ex -> ex.getMessage().contains("暂时不可用"))
.delayStrategy(attempt -> 1000L * attempt) // 线性退避
.onRetry(attempt ->
System.out.println("第 " + attempt + " 次重试..."))
.fallback(ex -> "降级结果")
.execute();
result.whenComplete((response, error) -> {
if (error != null) {
System.out.println("最终失败: " + error.getMessage());
} else {
System.out.println("最终结果: " + response);
}
});
// 等待异步操作完成
Thread.sleep(10000);
}
}
4. 注意事项
重试策略选择
- 网络超时：使用指数退避 + 随机抖动
- 服务限流：根据返回的等待时间重试
- 业务错误：根据具体错误码决定是否重试
避免的问题
- 无限重试：设置最大重试次数
- 资源耗尽：合理控制重试频率
- 雪崩效应：使用断路器模式配合重试
- 重复操作：确保操作的幂等性
监控和日志
// 添加重试监控
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);
}
}
具体业务场景选择合适的重试策略，提高系统的容错能力和稳定性。
以上就是Java中进行异步调用失败的解决方法详解的详细内容，更多关于Java异步调用的资料请关注IT俱乐部其它相关文章！
