Java 9 - 响应式流 ( Reactive Streams ) API 支持

雨点打透心脏的1/2处 2023-10-13 20:33 82阅读 0赞

package com.lfsun.java9study.flow;
    
    import java.util.concurrent.Flow;
    import java.util.concurrent.SubmissionPublisher;
    import java.util.concurrent.TimeUnit;
    
    public class ReactiveStreamsExample {
        
    
        public static void main(String[] args) {
        
            // 创建一个发布者
            SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
    
            // 创建一个订阅者
            Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {
        
                private Flow.Subscription subscription;
    
                @Override
                public void onSubscribe(Flow.Subscription subscription) {
        
                    this.subscription = subscription;
                    subscription.request(1); // 请求一个数据项
                }
    
                @Override
                public void onNext(String item) {
        
                    System.out.println("Received: " + item);
                    subscription.request(1); // 请求下一个数据项
                }
    
                @Override
                public void onError(Throwable throwable) {
        
                    System.err.println("Error occurred: " + throwable.getMessage());
                }
    
                @Override
                public void onComplete() {
        
                    System.out.println("Subscription completed.");
                }
            };
    
            // 订阅发布者
            publisher.subscribe(subscriber);
    
            // 发布数据
            publisher.submit("Data 1");
            publisher.submit("Data 2");
            publisher.submit("Data 3");
    
            // 关闭发布者
            publisher.close();
    
            try {
        
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
        
                throw new RuntimeException(e);
            }
        }
    }