如何在返回 Flux 的反应器调用序列中传递属性

Question

我正在将代码从命令式转变为反应式 java。

在命令式编程中，我们在 DTO/Context 中保存重要信息并随请求一起传递，有没有办法在使用 spring-webflux / project-reactor 的反应器范例中实现相同的目的？

换句话说::我怎样才能从多个非阻塞调用序列中维护所需的值。

例如，假设我有员工和部门 mongo 集合，并使用反应式 mongo 存储库从数据库中获取结果。

员工：

empId
状态：有效 |不活跃
empName
员工年龄
managerId
deptIds

部门：

deptId
部门名称

调用顺序

Flux<Employee> empFlux =  empRepository.findEmployees("ACTIVE"); // (1)
Flux<Department> deptFlux=  deptRepository.findDepartments(deptIds); // (2)


webclient (empId, managerId, deptId, deptName); // (3)

我在谷歌上搜索了一下，似乎我可以使用平面图来对这些流程进行排序，但是每次转换我都会丢失之前调用中的信息。

如上面的代码：

(1) : 获取活跃员工列表

(2) ：通过 id 平面图，我们获得了符合条件的部门列表，但丢失了有关员工的信息。

(3) 我需要使用有关员工、部门的信息对 webclient 进行反应性休息调用。

我的问题是如何在非阻塞通量和 Monos 序列中保留此信息。

Answer 1

对于这种情况，元组很有用：

empRepository.findEmployees("ACTIVE")
        .flatMap(emp -> deptRepository.findDepartments(emp.getDeptIds()).map(dept -> Tuples.of(emp,dept)))
        .flatMap(tuple -> webclient(tuple.getT1().getEmpId(), tuple.getT1().getMangagerId(), tuple.getT2().getDeptId(), tuple.getT2().getDeptName()));

IMO 这不是很可读，但对于一些官方运营商来说它也是 Reactor Way™，比如 zip。

Answer 2

您需要将该信息保存在一个单独的容器中，例如一个数组列表可访问的内容，并在以后需要时使用它。您可以尝试以下方法。

// Declare the arraylist to contain the return response object.
List<Object> returnedObjectList = new ArrayList<>();

Flux<Employee> empFlux =  empRepository.findEmployees("ACTIVE"); // (1)
Flux<Department> deptFlux=  deptRepository.findDepartments(deptIds); // (2)

return Flux.zip(empFlux, deptFlux)
         .flatMap(tuple -> {
             final String empId = tuple.getT1().getEmpId();
             final String managerId = tuple.getT1().getManagerId();

             final String deptId = tuple.getT2().getDeptId();
             final String deptName = tuple.getT2().getDeptName();

             Object returnedObject = webclient(empId, managerId, deptId, deptName); // (3)
             returnedObjectList.add(returnedObject);
             return Flux.just(returnedObject);

          });

Answer 3

您也可以像这样使用 zipwith 运算符：

Flux<WebClientReturn> infoFlux = empRepository.findEmployees("ACTIVE")
       .zipWith(deptRepository.findDepartments(deptIds), (employee, dept) -> {
          webclient (employee.getempId(), employee.getmanagerId(), dept.getdeptId(), dept.getdeptName());
        })

我认为这是最具描述性和最易读的方式。

如何在返回 Flux 的反应器调用序列中传递属性

how to pass attributes across sequence of reactor calls returing Flux

project-reactor

spring-webflux

员工：

部门：

调用顺序