使用Rx，如何获取每个唯一键的最新值？

import io.reactivex.Observable;
import io.reactivex.functions.Function;
import io.reactivex.observables.GroupedObservable;

public class RxUniqueKeyExample {
    public static void main(String[] args) {
        // 创建一个包含唯一键和值的数据源
        Observable<String> dataSource = Observable.just("key1:value1", "key2:value2", "key1:value3", "key3:value4");

        // 使用操作符将数据源转换为键值对的形式
        Observable<KeyValue> keyValueSource = dataSource.map(new Function<String, KeyValue>() {
            @Override
            public KeyValue apply(String s) throws Exception {
                String[] parts = s.split(":");
                return new KeyValue(parts[0], parts[1]);
            }
        });

        // 使用操作符对数据进行分组，按唯一键进行分组
        Observable<GroupedObservable<String, KeyValue>> groupedSource = keyValueSource.groupBy(new Function<KeyValue, String>() {
            @Override
            public String apply(KeyValue keyValue) throws Exception {
                return keyValue.getKey();
            }
        });

        // 对每个分组应用操作符，获取每个唯一键的最新值
        Observable<KeyValue> latestValues = groupedSource.flatMap(new Function<GroupedObservable<String, KeyValue>, Observable<KeyValue>>() {
            @Override
            public Observable<KeyValue> apply(GroupedObservable<String, KeyValue> groupedObservable) throws Exception {
                return groupedObservable.reduce(new KeyValueReducer()).toObservable();
            }
        });

        // 订阅最新值的可观察对象，处理每个唯一键的最新值
        latestValues.subscribe(new io.reactivex.Observer<KeyValue>() {
            @Override
            public void onSubscribe(io.reactivex.disposables.Disposable d) {
                // TODO: 处理订阅事件
            }

            @Override
            public void onNext(KeyValue keyValue) {
                // 处理每个唯一键的最新值
                System.out.println("Key: " + keyValue.getKey() + ", Latest Value: " + keyValue.getValue());
            }

            @Override
            public void onError(Throwable e) {
                // 处理错误事件
            }

            @Override
            public void onComplete() {
                // 处理完成事件
            }
        });
    }

    // 定义键值对类
    static class KeyValue {
        private String key;
        private String value;

        public KeyValue(String key, String value) {
            this.key = key;
            this.value = value;
        }

        public String getKey() {
            return key;
        }

        public String getValue() {
            return value;
        }
    }

    // 定义键值对归约器
    static class KeyValueReducer implements io.reactivex.functions.BiFunction<KeyValue, KeyValue, KeyValue> {
        @Override
        public KeyValue apply(KeyValue keyValue1, KeyValue keyValue2) throws Exception {
            // 比较两个键值对的时间戳，返回最新的键值对
            // 这里假设键值对的时间戳保存在值的字符串中，格式为 "value:timestamp"
            long timestamp1 = Long.parseLong(keyValue1.getValue().split(":")[1]);
            long timestamp2 = Long.parseLong(keyValue2.getValue().split(":")[1]);
            return timestamp1 > timestamp2 ? keyValue1 : keyValue2;
        }
    }
}