如何在Webflux中同步rest方法？

Question

问题在于，当同时调用

updateUser

时，会发生错误的余额更新。

2023-09-03T18:20:53.035+03:00  INFO 12668 --- [actor-tcp-nio-1] com.example.testsync.UserController      : current balance - 770.0, will be - 780.0
2023-09-03T18:20:53.043+03:00  INFO 12668 --- [actor-tcp-nio-1] com.example.testsync.UserController      : current balance - 770.0, will be - 780.0
2023-09-03T18:20:53.048+03:00  INFO 12668 --- [actor-tcp-nio-1] com.example.testsync.UserController      : current balance - 770.0, will be - 780.0

控制器看起来像这样：

@Slf4j
@RestController
@RequestMapping("/users")
@RequiredArgsConstructor
public class UserController {
    private final UserRepository userRepository;

    @PutMapping("/{userId}:updateBalance")
    public Mono<User> updateUser(@PathVariable("userId") Long userId, @RequestParam("diff") Double diff) {
        return userRepository.findById(userId)
                .flatMap(user -> {
                    log.info(String.format("current balance - %s, will be - %s", user.getBalance(), user.getBalance() + diff));
                    user.setBalance(user.getBalance() + diff);
                    return Mono.just(user);
                })
                .flatMap(userRepository::save);
    }
}

理论上，我需要按如下方式组织管道：

req_1 -> 1. userRepository.findByid(userId) 2. userRepository.save(user)
req_2 -> 1. userRepository.findById(userId) 2. userRepository.save(user)

同时调用 req_1 和 req_2。但我不知道如何识别:(

Answer 1

您可以将它们调度到同一线程，因此不需要同步（这是一个阻塞操作）。

Scheduler dbScheduler = Schedulers.newSingle("db");

public Mono<User> updateUser(...) {
    return userRepository.findById(userId)
            .map(user -> {
                log.info(String.format("current balance - %s, will be - %s", user.getBalance(), user.getBalance() + diff));
                user.setBalance(user.getBalance() + diff);
                return user;
            })
            .flatMap(userRepository::save)
            //
            // tell Reactor to schedule it to a single thread execution:
            .subscribeOn(dbScheduler);
}