如何在RxJava2中实现/使用maxBy？关于collect/reduce种子值的疑问

嘿，我完全懂你纠结的点——用reduce或者collect实现maxBy时，硬塞一个默认值确实不够优雅，尤其是你希望直接拿Observable发射的第一个元素当起始种子值的场景。下面给你几个实用的方案：

方案1：用无种子的reduce重载方法 #

RxJava里的reduce有个非常适合你的重载版本——不需要传入初始种子值！它会自动把Observable发射的第一个元素作为初始累积值，然后依次和后续元素对比更新。举个例子，假设你要根据User对象的age字段找最大的用户：

Observable<User> usersObservable = ...;

usersObservable
    .reduce((currentMax, nextUser) -> nextUser.getAge() > currentMax.getAge() ? nextUser : currentMax)
    .subscribe(maxUser -> {
        // 这里拿到的就是年龄最大的User
        // 注意：如果Observable是空的，这个订阅只会收到onComplete，不会触发onNext
    });

这个方案完美避开了默认值的问题，完全依赖流中的第一个元素作为起始点。如果需要处理空流的情况，你可以在后面链式调用defaultIfEmpty()（比如传入一个默认用户，但这是可选的，比一开始就硬塞种子值灵活多了）。

方案2：用collect配合原子引用容器 #

如果一定要用collect，可以借助AtomicReference来管理当前最大值，初始时设为null，第一个元素进来时自动把它设为初始值，后续元素再逐一对比更新：

Observable<User> usersObservable = ...;

usersObservable
    .collect(
        // 初始化容器：空的原子引用
        AtomicReference<User>::new,
        // 累积逻辑：对比并更新最大值
        (maxRef, currentUser) -> {
            User currentMax = maxRef.get();
            if (currentMax == null || currentUser.getAge() > currentMax.getAge()) {
                maxRef.set(currentUser);
            }
        }
    )
    .map(AtomicReference::get)
    .subscribe(maxUser -> {
        // 有元素时返回最大值，空流时返回null
    });

这种方式同样不需要提前指定默认种子值，完全由流中的元素决定初始值。

额外小提示：直接用内置的maxBy操作符 #

其实RxJava本身就提供了maxBy操作符，专门用来实现这个需求，内部已经帮你处理了种子值的问题，比自己手动用reduce/collect更简洁：

usersObservable
    .maxBy(User::getAge)
    .subscribe(maxUserMaybe -> {
        maxUserMaybe.ifPresent(user -> {
            // 处理找到的最大用户
        });
    });

它返回的是Maybe<T>，空流时会触发onComplete而不发射值，非空流则发射最大值，非常优雅。当然如果因为某些限制必须自己实现reduce/collect，前面的两个方案就很合适。

内容的提问来源于stack exchange，提问作者Alex Kokorin