如何通过Java Stream实现跨类方法间的数据传输？

一、原代码的问题 #

你写的伪代码存在几个关键问题：

• Stream是一次性消费的，一旦通过forEach等操作遍历后就无法复用，静态Stream变量会导致重复使用时抛出异常
• stream1.concat(stream2)返回的是新的Stream实例，但你没有将结果赋值给stream1，原stream1仍为初始的null
• 未指定泛型，应该使用Stream<SomeData>而非原始类型Stream

二、正确的Stream实现方式 #

由于不能直接在getDataFromUpstream中调用getData()，可以通过中间容器存储数据，按需生成Stream来实现传输：

方案1：全量数据的Stream获取 #

用线程安全的容器存储所有数据，每次需要时生成新的Stream，避免Stream一次性消费的限制：

Class1.java：

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;

public class Class1 {
    // 线程安全容器存储数据，避免并发修改问题
    private static final List<SomeData> DATA_STORE = new ArrayList<>();

    public void getDataFromUpstream(List<SomeData> data) {
        synchronized (DATA_STORE) {
            DATA_STORE.addAll(data);
        }
    }

    // 对外提供获取新Stream的方法
    public static Stream<SomeData> getFullStream() {
        synchronized (DATA_STORE) {
            return DATA_STORE.stream();
        }
    }
}

Class2.java：

public class Class2 {
    public void getData() {
        // 每次调用生成新Stream，处理当前所有数据
        Class1.getFullStream().forEach(data -> {
            // 替换为你的数据处理逻辑
            System.out.println("处理数据：" + data);
        });
    }
}

方案2：增量数据的Stream获取 #

如果只需要处理新增数据，而非重复处理全量数据，可以用阻塞队列实现流式增量处理：

Class1.java：

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Stream;

public class Class1 {
    private static final BlockingQueue<SomeData> DATA_QUEUE = new LinkedBlockingQueue<>();

    public void getDataFromUpstream(List<SomeData> data) {
        DATA_QUEUE.addAll(data);
    }

    // 生成阻塞式的增量Stream，等待新数据
    public static Stream<SomeData> getIncrementalStream() {
        return Stream.generate(() -> {
            try {
                return DATA_QUEUE.take();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return null;
            }
        }).takeWhile(data -> data != null);
    }
}

Class2.java：

public class Class2 {
    public void getData() {
        // 启动线程持续处理新增数据（避免阻塞主线程）
        new Thread(() -> {
            Class1.getIncrementalStream().forEach(data -> {
                // 替换为你的增量数据处理逻辑
                System.out.println("处理新增数据：" + data);
            });
        }).start();
    }
}

三、其他Java流式传输方案 #

1. Java原生Flow API（响应式流） #

用Java 9+内置的Flow API实现发布-订阅模式，原生支持背压控制，适合异步流式传输：

Class1.java（发布者）：

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

public class Class1 {
    private static final SubmissionPublisher<SomeData> PUBLISHER = new SubmissionPublisher<>();

    public void getDataFromUpstream(List<SomeData> data) {
        data.forEach(PUBLISHER::submit);
    }

    public static Flow.Publisher<SomeData> getPublisher() {
        return PUBLISHER;
    }
}

Class2.java（订阅者）：

import java.util.concurrent.Flow;

public class Class2 implements Flow.Subscriber<SomeData> {
    private Flow.Subscription subscription;

    public void getData() {
        Class1.getPublisher().subscribe(this);
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1); // 请求第一个数据
    }

    @Override
    public void onNext(SomeData item) {
        // 处理数据逻辑
        System.out.println("Flow API收到数据：" + item);
        subscription.request(1); // 请求下一个数据
    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
    }

    @Override
    public void onComplete() {
        System.out.println("数据传输完成");
    }
}

2. RxJava（第三方响应式库） #

如果允许引入第三方库，RxJava的Observable/Observer模式提供更灵活的流式操作：

Class1.java：

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.subjects.PublishSubject;

import java.util.List;

public class Class1 {
    private static final PublishSubject<SomeData> SUBJECT = PublishSubject.create();

    public void getDataFromUpstream(List<SomeData> data) {
        data.forEach(SUBJECT::onNext);
    }

    public static Observable<SomeData> getObservable() {
        return SUBJECT;
    }
}

Class2.java：

import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

public class Class2 {
    public void getData() {
        Class1.getObservable().subscribe(new Observer<SomeData>() {
            @Override
            public void onSubscribe(Disposable d) {
                // 可保存Disposable用于取消订阅
            }

            @Override
            public void onNext(SomeData someData) {
                // 处理数据逻辑
                System.out.println("RxJava收到数据：" + someData);
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {
                System.out.println("RxJava传输完成");
            }
        });
    }
}

内容的提问来源于stack exchange，提问作者Piyush Shrivastava