SpringCloud-Demo项目中的RxJava入门实践指南

2025-06-07 08:14:28作者：庞眉杨Will

前言

在分布式系统开发中，响应式编程已经成为处理异步数据流的重要范式。本文将基于SpringCloud-Demo项目中的RxJava示例，深入浅出地讲解RxJava的核心概念和基础用法。

环境准备

首先需要在项目中添加RxJava的依赖：

<dependencies>
    <dependency>
        <groupId>io.reactivex</groupId>
        <artifactId>rxjava</artifactId>
        <version>1.3.8</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
</dependencies>

RxJava核心概念

RxJava的核心是观察者模式，包含三个基本概念：

Observable：被观察者，负责产生事件
Observer/Subscriber：观察者，负责消费事件
Subscription：订阅关系

事件类型包括：

onNext()：正常事件
onCompleted()：事件队列完结
onError()：事件队列异常

基础示例

Hello World示例

@Test
public void helloWorld() {
    // 创建被观察者
    Observable<String> producer = Observable.create(subscriber -> {
        subscriber.onNext("hello world!");
        subscriber.onNext("apple!");
        subscriber.onNext("orange!");
        subscriber.onNext("exit");
        subscriber.onCompleted();
    });
    
    // 创建观察者
    Subscriber<String> consumer = new Subscriber<String>() {
        @Override
        public void onCompleted() {
            System.out.println("发送完成!");
        }
        
        @Override
        public void onError(Throwable e) {
            System.out.println("接受消息错误：" + e);
        }
        
        @Override
        public void onNext(String s) {
            System.out.println("收到消息：" + s);
        }
    };
    
    // 建立订阅关系
    producer.subscribe(consumer);
}

这个示例展示了RxJava最基本的用法：创建一个生产者Observable和一个消费者Subscriber，然后通过subscribe()方法建立订阅关系。

常用操作符

from操作符

将集合或数组转换为Observable：

@Test
public void from() {
    List<String> source = Arrays.asList("rxJava", "from", "exit");
    Observable.from(source).subscribe(s -> System.out.println("From: - " + s));
}

defer操作符

延迟创建Observable，直到有订阅者订阅时才创建：

@Test
public void defer() {
    List<String> source = Arrays.asList("rxJava", "defer", "exit");
    Observable<String> observable = Observable.defer(() -> {
        System.out.println("新建一个Observale生产者");
        return Observable.from(source);
    });
    
    observable.subscribe(s -> System.out.println("Defer1: - " + s));
    observable.subscribe(s -> System.out.println("Defer2: - " + s));
}

startWith操作符

在数据流开头插入数据：

@Test
public void startWith() {
    List<String> sources = Arrays.asList("rxJava", "startWith", "exit");
    Observable.from(sources)
        .startWith("Hello", "World")
        .subscribe(s -> System.out.println("Start with: - " + s));
}

filter操作符

过滤数据，类似于Java 8 Stream中的filter：

@Test
public void filter() {
    List<Integer> numbers = Arrays.asList(10, 2, 6, 3, 7, 5, 9, 8, 1);
    Observable.from(numbers)
        .filter(integer -> integer > 4)
        .subscribe(integer -> System.out.println("Filter : " + integer));
}

map操作符

转换数据：

@Test
public void map() {
    List<Integer> numbers = Arrays.asList(10, 2, 6, 3, 7, 5, 9, 8, 1);
    Observable.from(numbers)
        .map(integer -> "[map:" + integer + ']')
        .subscribe(o -> System.out.println("Map - > " + o.toString()));
}

flatMap操作符

一对多转换：

@Test
public void flatMap() {
    List<Integer> numbers = Arrays.asList(10, 2, 6, 3, 7, 5, 9, 8, 1);
    Observable.from(numbers)
        .flatMap(integer -> {
            List<Integer> integers = Arrays.asList(integer, integer * 5, integer * 10, integer * 20);
            return Observable.from(integers);
        })
        .subscribe(obj -> System.out.println("FlatMap: " + obj.toString()));
}

reduce操作符

聚合操作，类似递归：

@Test
public void reduce() {
    List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
    Observable.from(numbers)
        .reduce((integer, integer2) -> integer * integer2)
        .subscribe(integer -> System.out.println("1-9相乘结果：" + integer));
}

scan操作符

与reduce类似，但会输出每次计算结果：

@Test
public void scan() {
    List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
    Observable.from(numbers)
        .scan((integer, integer2) -> integer + integer2)
        .subscribe(integer -> System.out.println("Scan : " + integer));
}

高级应用：滑动窗口

windowCount操作符

固定大小的滑动窗口：

@Test
public void windowCount() {
    List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
    // 取相邻的2个数计算，然后跳跃2个事件
    Observable.from(numbers)
        .window(2, 2)
        .subscribe(integerObservable -> 
            integerObservable
                .reduce((integer, integer2) -> integer + integer2)
                .subscribe(integer -> System.out.println("Window ->" + integer))
        );
}

windowTime操作符

基于时间的滑动窗口：

@Test
public void windowTime() throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(1);
    Observable inputEventStream = Observable.create(subscriber -> 
        subscriber.onNext("我是生产者.........."));
    
    // 每个1s中输出一次
    inputEventStream.window(1000, TimeUnit.MILLISECONDS)
        .subscribe(o -> {
            Calendar calendar = Calendar.getInstance();
            int i = calendar.get(Calendar.SECOND);
            System.out.println(i + " -->执行");
        });
    
    countDownLatch.await();
}

综合滑动窗口示例

@Test
public void windowUltimateDemo() throws InterruptedException {
    // reduce 计算两个数字
    Func2<Integer, Integer, Integer> publicSum = (integer, integer2) -> integer + integer2;
    
    // 第一次统计
    Func1<Observable<Integer>, Observable<Integer>> reduceFunc = integerObservable -> 
        integerObservable.reduce(0, publicSum);
    
    // 程序计数等待器
    CountDownLatch countDownLatch = new CountDownLatch(1);
    
    // BehaviorSubject将数据流中的最新值推送给接受者
    BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create();
    
    behaviorSubject
        // 1s作为一个基本块
        .window(1000, TimeUnit.MILLISECONDS)
        // 将flatMap汇总平铺成一个事件
        .flatMap(reduceFunc)
        // 对这个对象2个发送，步长为1
        .window(2, 1)
        // 对窗口里面的进行求和
        .flatMap(reduceFunc)
        .subscribe(integer -> 
            System.out.println(Thread.currentThread().getName() + "-每秒统计输出结果：-> " + integer));
    
    for (int i = 0; i < 10000; i++) {
        behaviorSubject.onNext(i);
        // 200ms产生一个数据
        Thread.sleep(200);
    }
    countDownLatch.await();
}