Java异步流API怎么使用

Java异步流API使用指南

Java 9+ 引入了异步流API，主要通过Flow类实现响应式编程,下面是完整的使用指南：

核心接口

import java.util.concurrent.Flow.*;
// 四个核心接口
Publisher<T>    // 发布者
Subscriber<T>   // 订阅者
Subscription    // 订阅令牌
Processor<T,R>  // 处理器（既是发布者又是订阅者）

基础示例

简单发布-订阅模型

import java.util.concurrent.Flow.*;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;
public class FlowDemo {
    public static void main(String[] args) throws Exception {
        // 1. 创建发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        // 2. 创建订阅者
        Subscriber<String> subscriber = new Subscriber<>() {
            private Subscription subscription;
            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                System.out.println("订阅成功");
                subscription.request(1); // 请求第一个元素
            }
            @Override
            public void onNext(String item) {
                System.out.println("收到: " + item);
                subscription.request(1); // 处理完后再请求下一个
            }
            @Override
            public void onError(Throwable throwable) {
                System.err.println("错误: " + throwable.getMessage());
            }
            @Override
            public void onComplete() {
                System.out.println("所有数据接收完成");
            }
        };
        // 3. 订阅
        publisher.subscribe(subscriber);
        // 4. 发布数据
        publisher.submit("Hello");
        publisher.submit("World");
        publisher.submit("Java Flow API");
        // 5. 关闭发布者
        publisher.close();
        // 等待异步处理完成
        Thread.sleep(1000);
    }
}

背压控制（重要特性）

public class BackpressureDemo {
    public static void main(String[] args) throws Exception {
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
        Subscriber<Integer> slowSubscriber = new Subscriber<>() {
            private Subscription subscription;
            private int count = 0;
            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(3); // 只请求3个，实现背压
            }
            @Override
            public void onNext(Integer item) {
                count++;
                System.out.println("处理第" + count + "个: " + item);
                try {
                    Thread.sleep(500); // 模拟慢处理
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (count < 3) {
                    subscription.request(1);
                }
            }
            @Override
            public void onError(Throwable throwable) {
                System.err.println("错误: " + throwable.getMessage());
            }
            @Override
            public void onComplete() {
                System.out.println("处理完成");
            }
        };
        publisher.subscribe(slowSubscriber);
        // 快速发布
        for (int i = 1; i <= 10; i++) {
            int item = publisher.submit(i);
            System.out.println("发布 " + i + " 返回: " + item);
            // 返回负值表示缓冲区满
        }
        publisher.close();
        Thread.sleep(5000);
    }
}

处理器（Processor）示例

public class ProcessorExample {
    public static void main(String[] args) throws Exception {
        // 创建转换处理器：String -> Integer
        Processor<String, Integer> processor = new Processor<>() {
            private Subscriber<? super Integer> downstream;
            private Subscription upstream;
            @Override
            public void onSubscribe(Subscription subscription) {
                this.upstream = subscription;
                subscription.request(Long.MAX_VALUE);
            }
            @Override
            public void onNext(String item) {
                try {
                    int value = Integer.parseInt(item);
                    downstream.onNext(value * 2); // 转换并放大
                } catch (NumberFormatException e) {
                    downstream.onError(e);
                }
            }
            @Override
            public void onError(Throwable throwable) {
                downstream.onError(throwable);
            }
            @Override
            public void onComplete() {
                downstream.onComplete();
            }
            @Override
            public void subscribe(Subscriber<? super Integer> subscriber) {
                this.downstream = subscriber;
                subscriber.onSubscribe(new Subscription() {
                    @Override
                    public void request(long n) {
                        // 转发请求
                    }
                    @Override
                    public void cancel() {
                        upstream.cancel();
                    }
                });
            }
        };
        // 使用处理器
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        Subscriber<Integer> subscriber = new Subscriber<>() {
            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(Long.MAX_VALUE);
            }
            @Override
            public void onNext(Integer item) {
                System.out.println("处理后的值: " + item);
            }
            @Override
            public void onError(Throwable throwable) {
                System.err.println("处理错误: " + throwable.getMessage());
            }
            @Override
            public void onComplete() {
                System.out.println("所有处理完成");
            }
        };
        // 链式订阅
        publisher.subscribe(processor);
        processor.subscribe(subscriber);
        // 发布数据
        publisher.submit("10");
        publisher.submit("20");
        publisher.submit("30");
        publisher.close();
        Thread.sleep(1000);
    }
}

实用场景：数据转换处理

import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.Flow.*;
public class RealWorldExample {
    public static void main(String[] args) throws Exception {
        // 创建数据流处理管道
        // 1. 数据源发布者
        SubmissionPublisher<DataRecord> sourcePublisher = new SubmissionPublisher<>();
        // 2. 过滤处理器
        Processor<DataRecord, DataRecord> filterProcessor = newProcessor(
            subscriber -> new Subscription() {
                @Override
                public void request(long n) {
                    subscriber.onSubscribe(new Subscription() {
                        @Override
                        public void request(long n) {
                            // 实际项目中实现
                        }
                        @Override
                        public void cancel() {
                        }
                    });
                }
                @Override
                public void cancel() {
                }
            }
        );
        // 3. 转换处理器
        Processor<DataRecord, ProcessedData> transformProcessor = new Processor<>() {
            private Subscriber<? super ProcessedData> downstream;
            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(Long.MAX_VALUE);
            }
            @Override
            public void onNext(DataRecord record) {
                ProcessedData processed = new ProcessedData(
                    record.getId(),
                    record.getName().toUpperCase(),
                    record.getValue() * 2
                );
                downstream.onNext(processed);
            }
            @Override
            public void onError(Throwable throwable) {
                downstream.onError(throwable);
            }
            @Override
            public void onComplete() {
                downstream.onComplete();
            }
            @Override
            public void subscribe(Subscriber<? super ProcessedData> subscriber) {
                this.downstream = subscriber;
            }
        };
        // 4. 最终消费者
        Subscriber<ProcessedData> consumer = new Subscriber<>() {
            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(Long.MAX_VALUE);
            }
            @Override
            public void onNext(ProcessedData data) {
                System.out.println("处理结果: " + data);
            }
            @Override
            public void onError(Throwable throwable) {
                System.err.println("处理失败: " + throwable.getMessage());
            }
            @Override
            public void onComplete() {
                System.out.println("数据流处理完成");
            }
        };
        // 构建管道
        sourcePublisher.subscribe(transformProcessor);
        transformProcessor.subscribe(consumer);
        // 模拟数据生产
        sourcePublisher.submit(new DataRecord(1, "Alice", 100.0));
        sourcePublisher.submit(new DataRecord(2, "Bob", 200.0));
        sourcePublisher.submit(new DataRecord(3, "Charlie", 300.0));
        sourcePublisher.close();
        Thread.sleep(1000);
    }
    // 辅助类
    static class DataRecord {
        private int id;
        private String name;
        private double value;
        public DataRecord(int id, String name, double value) {
            this.id = id;
            this.name = name;
            this.value = value;
        }
        public int getId() { return id; }
        public String getName() { return name; }
        public double getValue() { return value; }
    }
    static class ProcessedData {
        private int id;
        private String name;
        private double value;
        public ProcessedData(int id, String name, double value) {
            this.id = id;
            this.name = name;
            this.value = value;
        }
        @Override
        public String toString() {
            return "ProcessedData{id=" + id + ", name='" + name + "', value=" + value + "}";
        }
    }
}

注意事项

1. 线程安全：SubmissionPublisher是线程安全的
2. 缓冲区大小：默认缓冲区大小为Flow.defaultBufferSize()
3. 错误处理：onError方法会终止订阅
4. 取消订阅：调用subscription.cancel()可以取消

与Reactive Streams兼容

// 与其他Reactive库配合
import org.reactivestreams.*;
// Java Flow API 与 Reactive Streams 规范兼容

这个API适合处理异步数据流、事件驱动编程和响应式应用场景。