当前位置：首页 > news >正文

Java Flow 编程：异步数据流介绍

news 来源：原创 2025/6/9 17:42:46

什么是Java Flow？

Java Flow是Java 9中引入的一个响应式编程API，它提供了一套标准的接口来处理异步数据流。这套API位于java.util.concurrent.Flow类中，包含了四个关键接口：Publisher、Subscriber、Subscription和Processor。

Flow API的设计受到了响应式流(Reactive Streams)规范的启发，旨在为Java提供一种标准的异步流处理方式，同时支持背压(backpressure)机制。

Flow的核心组件

1. Publisher（发布者）

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> subscriber);
}

Publisher是数据的生产者，负责生成数据项并发送给订阅者。一个Publisher可以有多个Subscriber。

2. Subscriber（订阅者）

public interface Subscriber<T> {
    void onSubscribe(Subscription subscription);
    void onNext(T item);
    void onError(Throwable throwable);
    void onComplete();
}

Subscriber是数据的消费者，通过回调方法接收来自Publisher的数据或事件通知。

3. Subscription（订阅）

public interface Subscription {
    void request(long n);
    void cancel();
}

Subscription代表了Publisher和Subscriber之间的一对一关系，用于控制数据流和背压。

4. Processor（处理器）

public interface Processor<T,R> extends Subscriber<T>, Publisher<R> {
}

Processor既是Subscriber又是Publisher，可以用于在数据流中执行转换操作。

背压(Backpressure)机制

Flow API的一个关键特性是内置的背压支持，这使得Subscriber可以控制它从Publisher接收数据的速度，防止被快速产生的数据淹没。

通过Subscription.request(long n)方法，Subscriber可以明确请求特定数量的数据项。Publisher必须尊重这个请求，不会发送超过请求数量的数据。

简单示例

下面是一个简单的Flow API使用示例：

import java.util.concurrent.Flow.*;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {
    public static void main(String[] args) {
        // 创建一个Publisher
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        
        // 创建Subscriber
        Subscriber<String> subscriber = new Subscriber<>() {
            private Subscription subscription;
            
            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1); // 请求第一个数据项
            }
            
            @Override
            public void onNext(String item) {
                System.out.println("接收到: " + item);
                subscription.request(1); // 请求下一个数据项
            }
            
            @Override
            public void onError(Throwable throwable) {
                throwable.printStackTrace();
            }
            
            @Override
            public void onComplete() {
                System.out.println("处理完成");
            }
        };
        
        // 订阅
        publisher.subscribe(subscriber);
        
        // 发布数据
        publisher.submit("数据1");
        publisher.submit("数据2");
        publisher.submit("数据3");
        
        // 关闭Publisher
        publisher.close();
    }
}