如何连锁Reactor订户

Question

我有一个现有的接口链，我想作为一个反应器运行，而不是管理我自己的线程和队列

public interface UserLookupService {
    public User lookup(String id);
}
public interface UsersHandler {
    public void handle(List<User> users>);
}

UserLookupService userSvc = ...;
UsersHandler usersHandler = ...

// Works well to lookup users in parallel. 
Flux.just("userA", "userB", "userC")
    .parallel(2)
    .runOn(Schedulers.parallel())
    .subscribe(str -> {
        userSvc.lookup(str);
    });

我如何链接该结果，使其调用UsersHandler与批量的User？

Answer 1

订阅某些东西会触发链，所以你通常不能“链”订阅者，他们是链中的最后一件事。

想想看，如果这样，你设置了你的反应式管道，当你订阅时，你触发管道启动，链就会产生一个结果。

在web服务器中，< code>subscriber通常是调用客户端，当客户端< code > subscribe 时，他会触发将发布数据的服务器中的事件链。

通量有点像1到n个单声道的列表。可以说，单声道/通量中的每个对象都有许多“状态”。这些是成功，错误，取消，下一步，已完成等。

当Mono/Flux内部进入成功状态时，它将发出其中的值。当mono中的某些问题得到解决时，Mono通常会进入成功。

当你声明Flux.just（"userA"，"userB"，"userC"）时，你基本上是在要求通量来解析你输入到它的输入。放置一个字符串会立即解析，所以通量会进入成功状态，一旦有东西订阅就开始发出字符串。所以你所要做的就是在某人订阅后声明你想要发生的链。

这可以通过几种不同的方式完成，当你想做某事并更改值时，就像你想从字符串到我们通常使用map的用户一样。

如果我们只想对每个对象做一些事情而不返回任何东西，我们可以使用doOnNext。

Flux.just("userA", "userB", "userC")
            .parallel(2)
            .runOn(Schedulers.parallel())
            .map(userString -> {
                return lookupService.lookup(userString);
            })
            .doOnNext(user -> {
                // if you want to do something on each user
                // will return void so if you want to log something
                // or handle each user
            }).subscribe();

订阅应该是链条中的最后一环。