Java 11 - 响应式流和反应式编程

水深无声 2023-10-13 19:04 48阅读 0赞

## 1、订阅者 ##

package com.lfsun.java11study.juc;
    
    import java.util.concurrent.Flow;
    
    /**
     * 代表了一个订阅者，用于接收发布者发出的数据项
     * @param <T>
     */
    class SimpleSubscriber<T> implements Flow.Subscriber<T> {
        
        private Flow.Subscription subscription;
    
        @Override
        public void onSubscribe(Flow.Subscription subscription) {
        
            this.subscription = subscription;
            subscription.request(1); // 请求一个数据项
        }
    
        @Override
        public void onNext(T item) {
        
            System.out.println("Received: " + item);
            subscription.request(1); // 继续请求下一个数据项
        }
    
        @Override
        public void onError(Throwable throwable) {
        
            throwable.printStackTrace();
        }
    
        @Override
        public void onComplete() {
        
            System.out.println("Completed");
        }
    }

## 2、发布者 ##

package com.lfsun.java11study.juc;
    
    import java.util.concurrent.Flow;
    import java.util.concurrent.SubmissionPublisher;
    
    /**
     * 用于发布数据项
     */
    public class ReactiveStreamsDemo {
        
        public static void main(String[] args) {
        
            // 创建一个 SubmissionPublisher 对象
            SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
    
            // 创建一个订阅者
            Flow.Subscriber<String> subscriber = new SimpleSubscriber<>();
    
            // 订阅订阅者
            publisher.subscribe(subscriber);
    
            // 发布数据项
            publisher.submit("Hello");
            publisher.submit("Reactive");
            publisher.submit("Streams");
    
            // 添加一个延迟以保持程序运行并处理数据项
            try {
        
                Thread.sleep(1000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
    
            // 完成发布
            publisher.close();
        }
    }