说明

ReBus 是一个模仿 RxBus 写的，基于 reactor3 的事件总线

RxBus 一般是 Android 端使用的，
ReBus 一般是后端使用的
RxBus 参考链接： https://github.com/relengxing/RxBus
ReBus 源码链接： https://github.com/relengxing/ReBus

推荐直接看 RxBus 的解析，这边只贴代码，因为思想是一模一样的，只是底层的包不一样，封装完后用法也是一样的。

源码解析

ReBus 代码

package com.relenxing.config;
import com.relenxing.domain.Event;
import reactor.core.publisher.Flux;
import reactor.core.publisher.ReplayProcessor;
public class ReBus {

    /**
     * 事件总线的核心。
     */
    private final ReplayProcessor<Event> bus;

    /**
     * 构造函数，初始化
     */
    private ReBus() {
        bus = ReplayProcessor.create();
    }

    /**
     * 单例模式
     */
    public static ReBus getDefault() {
        return HelperHolder.instance;
    }

    /**
     * 延迟初始化，这里是利用了 Java 的语言特性，内部类只有在使用的时候，才会去加载，
     * 从而初始化内部静态变量。关于线程安全，这是 Java 运行环境自动给你保证的，
     * 在加载的时候，会自动隐形的同步。在访问对象的时候，
     * 不需要同步 Java 虚拟机又会自动给你取消同步，所以效率非常高。
     */
    private static class HelperHolder {
        static final ReBus instance = new ReBus();
    }


    /**
     * 发送普通事件
    */
    public void post(Event event){
        bus.onNext(event);
    }

    public  Flux<Event> on(String type){
        return bus.filter(e -> e.getEventName().equals(type));
    }
}

Event 代码：

package com.relenxing.domain;

import java.io.Serializable;

public class Event<T> implements Serializable {

    private static final long serialVersionUID = 1L;

    private String eventName;

    private T event;


    public Event() {
    }

    public Event(String eventName, T event) {
        this.eventName = eventName;
        this.event = event;
    }

    public String getEventName() {
        return eventName;
    }

    public void setEventName(String eventName) {
        this.eventName = eventName;
    }

    public T getEvent() {
        return event;
    }

    public void setEvent(T event) {
        this.event = event;
    }
}

demo

package com.relenxing;
public class Main {

    public static void main(String[] args) {
       ReBus.getDefault().post(new Event<>("123", "123"));
       ReBus.getDefault().post(new Event<>("456", "swdw"));
       ReBus.getDefault().post(new Event<>("123", "312"));
       ReBus.getDefault().post(new Event<>("456", "2f2"));
       ReBus.getDefault().post(new Event<>("123", "12f"));

       ReBus.getDefault().on("123").map(Event::getEvent).subscribe(System.out::println);
       ReBus.getDefault().on("456").map(Event::getEvent).subscribe(System.out::println);
    }
}