基于 WebFlux 的监听器

WebFlux 是原生的发布订阅工具，可以很方便的构建事件总线。下面是一个监听数据变动的监听器：

package com.example.demo;

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * 数据监听器
 *
 * @author <a href="mailto:pushu@2dfire.com">朴树</a>
 * @date 2019-07-01 17:32
 */
public class ReactorDataMonitor {
    private static final Map<Class, FluxSink> handlers = new ConcurrentHashMap<>();

    /**
     * 监控指定类型的数据
     *
     * @param clz     数据类型
     * @param handler 数据消费方式
     */
    public static void monitor(Class clz, Consumer handler) {
        Flux<Object> objectFlux = Flux.create(sink -> {
            handlers.put(clz, sink);
            sink.onCancel(() -> handlers.remove(clz));
        }, FluxSink.OverflowStrategy.LATEST);

        objectFlux.subscribe(handler);
    }

    /**
     * 取消监控数据
     *
     * @param clz 数据类型
     */
    public static void unMonitor(Class clz) {
        handlers.remove(clz);
    }

    /**
     * 发布数据
     *
     * @param object
     */
    public static void publish(Object object) {
        handlers.forEach((key, value) -> {
            if (key.equals(object.getClass())) {
                value.next(object);
            }
        });
    }

}

以上代码中FluxSink是一个可以持续发布数据的数据源。