21. java9 的发布订阅

ゝ一纸荒年。 2024-04-17 21:45 46阅读 0赞

> java 多线程系列文章列表, 请查看目录: [《java 多线程学习笔记》][java]

## 1. java9 的发布订阅 ##

java9 新增了一个发布-订阅框架, 该框架是基于异步响应流的. 之前开发发布订阅通常都会借助于中间件, 如mq, redis 等, java9之后, 对于简单的发布订阅系统, jdk便直接可以满足.

### 1.1 相关API ###

*  Flow.Pulisher: 消息的生产者, 发布者
 *  Flow.Subscriber: 数据的订消费者, 订阅者
 *  Flow.Subscription: 订阅者与发布者直接的连接通道, 订阅者可通过request方法获取数据项, 也可通过cancel 方法取消订阅
 *  Flow.Processor: 数据处理器, 可同时被发布者和订阅者使用.

### 1.2 消费者接口-Flow.Subscriber: ###

自定义消费者时, 需要实现Flow.Subscriber 接口, 并重写以下四个方法:

*  public void onSubscribe(Flow.Subscription subscription): 订阅者注册成功时, 自动触发. 当发布者发布一条消息时, 才会触发.
 *  public void onNext(T item): 获取到数据时, 触发该方法
 *  public void onError(Throwable throwable): 订阅失败时, 触发
 *  public void onComplete(): 订阅结束时触发. 当发布者关闭时, 订阅结束.

## 2. 发布订阅测试 ##

### 2.1 订阅者 ###

public class MySubscriber<T> implements Flow.Subscriber<T> { 
    
        // 发布者
        private Flow.Subscription subscription;
    
        // 名称
        private String name;
    
        public MySubscriber(String name) { 
            this.name = name;
        }
    
        @Override
        public void onSubscribe(Flow.Subscription subscription) { 
            this.subscription = subscription;
    
            System.out.println(this.name + "-注册成功");
    
            // 开始请求
            subscription.request(1);
        }
    
        @Override
        public void onNext(T item) { 
            System.out.println(this.name + " 获取到数据:" + item);
    
            subscription.request(1);
        }
    
        @Override
        public void onError(Throwable throwable) { 
            throwable.printStackTrace();
        }
    
        @Override
        public void onComplete() { 
            System.out.println(this.name + "订阅结束");
        }
    }

### 2.2 生产者测试 ###

public static void main(String[] args) { 
    
        // 创建发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
    
        // 创建订阅者
        MySubscriber<String> mySubscriber = new MySubscriber<>("张三");
    
        // 注册第一个订阅者
        publisher.subscribe(mySubscriber);
    
        System.out.println("1 秒后开始发布数据....");
        ThreadUtil.sleep(1);
    
        // 3秒后注册第二个订阅者
        new Thread(){ 
            @Override
            public void run() { 
                ThreadUtil.sleep(3);
                MySubscriber<String> mySubscriber2 = new MySubscriber<>("李四");
                publisher.subscribe(mySubscriber2);
            }
        }.start();
    
        // 发布数据
        List<String> list = List.of("a", "b", "c", "d", "e", "f");
        list.forEach(im -> { 
            publisher.submit(im);
            ThreadUtil.sleep(1);
        });
    
        // 发布者频道关闭, 通知订阅者结束.
        publisher.close();
    }

### 2.3 控制输出 ###

*  当发布第一条消息之后, 才打印了注册成功. 说明当生产者开始发布消息时, 才建立好了连接
 *  当生产者发布消息时, 所有订阅者均能得到消息
 *  当生产者关闭时, 所有订阅者均触发onComplete() 方法
 *  订阅者只接收开始订阅之后的消息, 不接收历史消息.

1 秒后开始发布数据....
    张三-注册成功
    张三 获取到数据:a
    张三 获取到数据:b
    张三 获取到数据:c
    李四-注册成功
    李四 获取到数据:d
    张三 获取到数据:d
    李四 获取到数据:e
    张三 获取到数据:e
    张三 获取到数据:f
    李四 获取到数据:f
    张三订阅结束
    李四订阅结束

[java]: https://blog.csdn.net/zongf0504/article/details/100186060