在微服务开发中，经常需要按顺序调用多个下游服务，比如先调用用户服务获取用户信息，再调用订单服务查询该用户的订单，最后调用物流服务获取订单的物流状态。如果使用传统的同步阻塞方式，线程会在等待服务响应时被占用，当并发量较高时会导致系统吞吐量下降。Mutiny 作为轻量级的响应式编程库，提供了声明式的异步操作编排能力，非常适合处理这类顺序调用场景。
Mutiny 核心类型介绍
Mutiny 有两个核心类型，分别是 Uni 和 Multi，其中 Uni 用于表示最多发射一个元素的异步操作，非常适合单个服务的调用场景。
- Uni：代表一个异步的、可能失败的单结果操作，比如一次 HTTP 请求、一次数据库查询都可以用 Uni 来封装。
- 链式调用：Uni 提供了 map、flatMap、onFailure 等方法，可以方便地串联多个操作，实现服务顺序调用。
基础服务顺序调用实现
假设我们有三个服务，分别是 UserService、OrderService、LogisticsService，需要按顺序调用，每个服务的调用都返回 Uni 类型的结果。
服务接口定义
首先定义三个服务的接口，每个接口的方法都返回 Uni 类型：
// 用户服务接口
public interface UserService {
Uni<User> getUserById(String userId);
}
// 订单服务接口
public interface OrderService {
Uni<List<Order>> getOrdersByUserId(String userId);
}
// 物流服务接口
public interface LogisticsService {
Uni<List<Logistics>> getLogisticsByOrderIds(List<String> orderIds);
}
// 用户实体类
public class User {
private String id;
private String name;
// 省略getter、setter
}
// 订单实体类
public class Order {
private String id;
private String userId;
// 省略getter、setter
}
// 物流实体类
public class Logistics {
private String orderId;
private String status;
// 省略getter、setter
}
顺序调用实现
使用 Uni 的 flatMap 方法可以串联多个异步操作，前一个操作的结果可以作为后一个操作的输入：
public class ServiceSequenceCaller {
private final UserService userService;
private final OrderService orderService;
private final LogisticsService logisticsService;
// 构造方法注入服务依赖
public ServiceSequenceCaller(UserService userService, OrderService orderService, LogisticsService logisticsService) {
this.userService = userService;
this.orderService = orderService;
this.logisticsService = logisticsService;
}
/**
* 按顺序调用三个服务，返回最终的物流信息
*/
public Uni<List<Logistics>> callServicesInSequence(String userId) {
return userService.getUserById(userId)
// 第一步：获取用户信息，失败则直接返回错误
.onFailure().recoverWithUni(failure -> {
System.err.println("获取用户失败：" + failure.getMessage());
return Uni.createFrom().failure(failure);
})
// 第二步：用用户ID查询订单，flatMap用于将当前Uni的结果转换为新的Uni
.flatMap(user -> orderService.getOrdersByUserId(user.getId()))
.onFailure().recoverWithUni(failure -> {
System.err.println("获取订单失败：" + failure.getMessage());
return Uni.createFrom().failure(failure);
})
// 第三步：从订单中提取订单ID，查询物流信息
.flatMap(orders -> {
List<String> orderIds = orders.stream()
.map(Order::getId)
.collect(java.util.stream.Collectors.toList());
return logisticsService.getLogisticsByOrderIds(orderIds);
})
.onFailure().recoverWithUni(failure -> {
System.err.println("获取物流失败：" + failure.getMessage());
return Uni.createFrom().failure(failure);
});
}
}
异常处理与结果包装
在实际场景中，我们可能需要对异常进行统一处理，或者将多个服务的结果包装成一个统一的响应对象返回。
统一异常返回
可以定义一个统一的响应类，将成功和失败的结果都封装进去：
public class ServiceResponse<T> {
private boolean success;
private T data;
private String errorMsg;
public ServiceResponse(boolean success, T data, String errorMsg) {
this.success = success;
this.data = data;
this.errorMsg = errorMsg;
}
public static <T> ServiceResponse<T> success(T data) {
return new ServiceResponse<>(true, data, null);
}
public static <T> ServiceResponse<T> fail(String errorMsg) {
return new ServiceResponse<>(false, null, errorMsg);
}
// 省略getter、setter
}
修改调用方法，将异常转换为统一的失败响应：
public Uni<ServiceResponse<List<Logistics>>> callServicesWithResponse(String userId) {
return userService.getUserById(userId)
.flatMap(user -> orderService.getOrdersByUserId(user.getId()))
.flatMap(orders -> {
List<String> orderIds = orders.stream()
.map(Order::getId)
.collect(java.util.stream.Collectors.toList());
return logisticsService.getLogisticsByOrderIds(orderIds);
})
// 成功时包装为成功响应
.map(ServiceResponse::success)
// 捕获所有异常，包装为失败响应
.onFailure().recoverWithUni(failure ->
Uni.createFrom().item(ServiceResponse.fail(failure.getMessage()))
);
}
调用示例
在实际使用中，可以订阅 Uni 的结果来处理最终返回值：
public class Main {
public static void main(String[] args) {
// 假设已经初始化了三个服务的实例
UserService userService = new UserService() {
@Override
public Uni<User> getUserById(String userId) {
User user = new User();
user.setId(userId);
user.setName("测试用户");
return Uni.createFrom().item(user);
}
};
OrderService orderService = new OrderService() {
@Override
public Uni<List<Order>> getOrdersByUserId(String userId) {
Order order = new Order();
order.setId("order_001");
order.setUserId(userId);
return Uni.createFrom().item(java.util.Collections.singletonList(order));
}
};
LogisticsService logisticsService = new LogisticsService() {
@Override
public Uni<List<Logistics>> getLogisticsByOrderIds(List<String> orderIds) {
Logistics logistics = new Logistics();
logistics.setOrderId(orderIds.get(0));
logistics.setStatus("已发货");
return Uni.createFrom().item(java.util.Collections.singletonList(logistics));
}
};
ServiceSequenceCaller caller = new ServiceSequenceCaller(userService, orderService, logisticsService);
caller.callServicesWithResponse("user_001")
.subscribe().with(
response -> {
if (response.isSuccess()) {
System.out.println("调用成功，物流信息：" + response.getData());
} else {
System.out.println("调用失败：" + response.getErrorMsg());
}
},
failure -> System.err.println("订阅失败：" + failure.getMessage())
);
}
}
注意事项
- flatMap 和 map 的区别：map 用于同步转换结果，不会返回 Uni 类型；flatMap 用于异步转换，需要返回 Uni 类型，服务调用场景必须使用 flatMap。
- 线程模型：Mutiny 默认在非阻塞线程上执行操作，不要在回调中执行耗时的阻塞操作，否则会影响响应式系统的性能。
- 资源释放：如果服务调用涉及资源连接，需要在 Uni 的操作中正确处理资源释放逻辑，避免资源泄漏。