响应式编程中的阻塞陷阱与错误处理
在传统的命令式编程中，try-catch-finally结构是处理异常和确保资源清理的标准范式。finally块中的代码无论是否发生异常都会执行，常用于关闭文件句柄、释放锁或保存状态。然而，在project reactor等响应式框架中，直接套用这种模式，尤其是在finally块中执行阻塞操作，将严重破坏响应流的非阻塞特性，导致性能瓶颈甚至死锁。
响应式编程的核心在于构建异步、非阻塞的数据流。当流中出现错误时，它会发出一个错误信号，而不是像命令式代码那样抛出异常并中断线程。因此，在Reactor中，我们不应直接抛出运行时异常，而应使用Mono.error()或Flux.error()来发出错误信号。
Reactor提供了丰富的操作符来处理流中的错误信号，这些操作符允许我们以非阻塞的方式响应错误：
- doOnError(Consumer super Throwable> onError): 用于执行副作用操作，例如日志记录。它不会改变流的错误信号，错误会继续向下游传播。
- onErrorResume(Function super Throwable, ? extends Publisher extends T>> fallback): 当上游发出错误信号时，提供一个替代的响应式流（Mono或Flux）来继续处理。这对于实现错误恢复或提供默认值非常有用。
- onErrorMap(Function super Throwable, ? extends Throwable> errorMapper): 用于将一种类型的错误转换为另一种类型的错误，然后将新错误向下游传播。
- 避免使用 onErrorContinue: 这是一个特殊的操作符，它允许在发生错误时跳过有问题的元素并继续处理流中的其他元素。但在大多数业务场景中，错误通常意味着整个操作的失败，继续处理可能导致数据不一致或逻辑混乱，因此应谨慎使用或避免。
模拟“finally”逻辑的响应式实现
在命令式代码中，finally块的目的是无论成功或失败都执行特定逻辑。在Reactor中，这意味着我们需要将这些逻辑嵌入到流的成功路径和错误路径中。
考虑以下原始的命令式逻辑：
public Monoprocess(Request request) { var existingData = repository.find(request.getId()); // 查找现有数据 if (existingData != null) { if (existingData.getState() != pending) { throw new RuntimeException("test"); // 状态不符则抛异常 } } else { existingData = repository.save(convertToData(request)); // 无数据则保存新数据 } try { var response = hitAPI(existingData); // 调用外部API } catch(ServerException serverException) { log.error(""); throw serverException; // API调用失败则抛异常 } finally { repository.save(existingData); // 无论成功失败，都保存数据 } return convertToResponse(existingData, response); // 转换响应 }
这段代码存在多个阻塞操作，并且finally块中的repository.save(existingData)也是阻塞的。为了将其转换为响应式代码，并模拟finally的行为，我们需要将保存操作集成到流的成功和失败路径中。
以下是经过优化和修正的Reactor响应式实现：
import reactor.core.publisher.Mono;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// 假设的依赖和实体
class Request { String getId() { return null; } }
class Response {}
class Data { Object getState() { return null; } } // 假设有getState方法
enum State { pending, completed } // 假设有pending状态
class ServerException extends RuntimeException {}
// 假设的Repository接口（返回Mono）
interface ReactiveRepository {
Mono find(String id);
Mono save(Data data);
}
public class ReactiveProcessService {
private static final Logger log = LoggerFactory.getLogger(ReactiveProcessService.class);
private final ReactiveRepository repository;
public ReactiveProcessService(ReactiveRepository repository) {
this.repository = repository;
}
private Data convertToData(Request request) { /* 转换逻辑 */ return new Data(); }
private Response convertToResponse(Data data, Object response) { /* 转换逻辑 */ return new Response(); }
private Object hitAPI(Data data) throws ServerException { /* 模拟外部API调用 */ return new Object(); }
public Mono process(Request request) {
return repository.find(request.getId())
.flatMap(existingData -> {
// 如果找到现有数据
if (existingData.getState() != State.pending) {
// 如果状态不是pending，则发出错误信号
return Mono.error(new RuntimeException("Data state is not pending."));
} else {
// 如果状态是pending，则继续使用现有数据
return Mono.just(existingData);
}
})
.switchIfEmpty(Mono.defer(() -> repository.save(convertToData(request)))) // 如果未找到数据，则保存新数据
.flatMap(existingData -> Mono
// 包装可能阻塞的API调用，使其在响应式流中执行
.fromCallable(() -> hitAPI(existingData))
// 捕获ServerException，记录日志，但不中断流（错误信号会继续传播）
.doOnError(ServerException.class, throwable -> log.error("API call failed: {}", throwable.getMessage(), throwable))
// 错误处理路径：如果API调用失败，先保存数据，再重新发出错误信号
.onErrorResume(throwable ->
repository.save(existingData) // 执行“finally”逻辑：保存数据
.then(Mono.error(throwable)) // 然后重新发出原始错误信号
)
// 成功处理路径：如果API调用成功，先保存数据，再转换响应
.flatMap(apiResponse ->
repository.save(existingData) // 执行“finally”逻辑：保存数据
.map(updatedExistingData -> convertToResponse(updatedExistingData, apiResponse))
)
);
}
} 代码解析：
- repository.find(request.getId()): 开始流，尝试查找现有数据。
-
第一个 flatMap:
- 如果find操作找到了数据（existingData），则进入此flatMap。
- 检查existingData的状态。如果不是pending，则通过Mono.error()发出一个错误信号，流将转向错误处理路径。
- 如果状态是pending，则通过Mono.just(existingData)将现有数据向下游传递。
-
switchIfEmpty(Mono.defer(() -> repository.save(convertToData(request)))):
- 如果repository.find返回Mono.empty()（即未找到数据），则switchIfEmpty会被激活。
- Mono.defer()用于延迟执行repository.save，确保只有在find确实为空时才执行保存新数据的操作。
- repository.save(convertToData(request))会保存新数据并将其向下游传递。
-
第二个 flatMap: 此时existingData已被确定（要么是找到的现有数据，要么是新保存的数据）。
- Mono.fromCallable(() -> hitAPI(existingData)): 这是一个关键步骤。hitAPI可能是一个传统的、潜在阻塞的方法。fromCallable将其包装成一个Mono，使其在订阅时执行，并且可以在合适的调度器上运行，从而避免阻塞主线程。
- doOnError(ServerException.class, ...): 这是一个副作用操作符。如果hitAPI抛出ServerException，这里会捕获并记录日志。错误信号会继续向下游传播。
-
onErrorResume(throwable -> ...) (错误处理路径): 如果上游（hitAPI或之前的操作）发出任何错误信号，此操作符将被激活。
- repository.save(existingData): 这是模拟finally行为的关键部分。在错误发生时，我们首先执行保存操作。
- .then(Mono.error(throwable)): then操作符用于在完成前一个Mono（这里是save操作）后，忽略其结果并执行下一个Mono。这里我们在保存完成后，重新发出原始的错误信号，确保错误继续向下游传播，通知调用者操作失败。
-
flatMap(apiResponse -> ...) (成功处理路径): 如果hitAPI成功返回apiResponse，此操作符将被激活。
- repository.save(existingData): 同样是模拟finally行为的关键部分。在成功时，我们也执行保存操作。
- .map(updatedExistingData -> convertToResponse(updatedExistingData, apiResponse)): 保存成功后，将更新后的existingData和apiResponse转换为最终的Response并向下游传递。
注意事项与总结
- 响应式仓库是前提: 上述代码假设repository.find和repository.save方法返回Mono，即它们本身就是非阻塞的响应式操作。如果你的仓库层是阻塞的（例如传统的JPA），你需要使用Mono.fromCallable()或Mono.just().subscribeOn(Schedulers.boundedElastic())等方式将其包装起来，并确保在合适的调度器上执行。
- finally逻辑的复制: 在响应式编程中，finally块的逻辑（例如这里的repository.save(existingData)）通常需要在成功路径和错误路径中分别实现。虽然这看起来是代码复制，但它是确保非阻塞和正确处理流的必要方式。
- 避免在flatMap中直接抛出异常: 始终使用Mono.error()来发出错误信号，而不是throw new RuntimeException()。
- Mono.defer的妙用: 在switchIfEmpty等场景中，使用Mono.defer可以确保懒加载，即只有当实际需要时才创建并执行内部的Mono。
通过上述方法，我们成功地将传统的try-catch-finally结构转换为Reactor流的非阻塞范式，确保了在成功和失败情况下都能执行必要的副作用操作，同时保持了响应式应用程序的性能和响应性。
