Guava EventBus 使用介绍及自实现详解

发表于 | 分类 |
| 全文总计: 字
Guava EventBus，内存级别消息发布/订阅模型

使用 guava 工具，先在 pom 中引入依赖：https://mvnrepository.com/artifact/com.google.guava/guava/

1
2
3
4
5
<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>29.0-jre</version>
</dependency>

Guava EventBus 的使用

简单使用

先编写监听者：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import com.google.common.eventbus.Subscribe;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:08
 * @description:
 */
public class GoogleEventBusListener {

    @Subscribe
    public void method1(String event) {
        System.out.println("GoogleEventBusListener.method1, event = " + event);
    }

    @Subscribe
    public void method2(Integer event) {
        System.out.println("GoogleEventBusListener.method2, event = " + event);
    }
}

再创建 EventBus 对象，将上述监听者注册到 EventBus 对象实例中，调用 post() 发送消息即可：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import com.google.common.eventbus.EventBus;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:08
 * @description: EventBus 简单示例
 */
public class SimpleEventBusExample {

    public static void main(String[] args) {
        EventBus eventBus = new EventBus();

        // 监听器注册到 eventBus
        eventBus.register(new GoogleEventBusListener());

        // 发送消息
        eventBus.post("this message from eventBus");
        eventBus.post(200);
    }

}

EventBus 构造函数中，如果不显示指定名称，则使用：default 作为默认名称。

注意：消息类型必须是包装类型，不能是基本数据类型。

多个监听者存在继承关系

当只注册监听器中的父类中也监听了某些消息的时候，父类的监听方法也会执行。如果父类的监听方法被子类覆盖监听，则只执行子类监听器的方法。

类监听器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import com.google.common.eventbus.Subscribe;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:18
 * @description: 父类监听器
 */
public class FatherListener {

    @Subscribe
    public void accept(String event) {
        System.out.println("FatherListener.accept, event = " + event);
    }

    @Subscribe
    public void consume(String event) {
        System.out.println("FatherListener.consume, event = " + event);
    }
}

子类监听器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import com.google.common.eventbus.Subscribe;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:19
 * @description: 子类监听器
 */
public class ChildListener extends FatherListener {

    @Subscribe
    @Override
    public void accept(String event) {
        System.out.println("ChildListener.accept, event = " + event);
    }

    @Subscribe
    public void consumeTask(String event) {
        System.out.println("ChildListener.consumeTask, event = " + event);
    }
}

注册监听器并发送消息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import com.google.common.eventbus.EventBus;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:21
 * @description: 多个监听器监听消息
 * 当只注册监听器中的父类中也监听了某些消息的时候，父类的监听方法也会执行。
 * 如果父类的监听方法被子类覆盖监听，则只执行子类监听器的方法
 *
 */
public class MultipleListenerExample {

    public static void main(String[] args) {
        EventBus eventBus = new EventBus();
        // 监听器注册到 eventBus
        eventBus.register(new ChildListener());

        // 发送消息
        eventBus.post("this message from eventBus");
    }
}

日志输出：

1
2
3
ChildListener.accept, event = this message from eventBus
ChildListener.consumeTask, event = this message from eventBus
FatherListener.consume, event = this message from eventBus

多个事件存在继承关系

event 如果有父类，此时某个监听器监听的event是这个基类，那么消息是这个子类的时候，该监听父类的监听器方法也会执行。

父类事件对象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import com.google.gson.Gson;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:33
 * @description: 父类事件对象
 */
public class FruitEvent {

    String name;

    public FruitEvent() {

    }

    public FruitEvent(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return new Gson().toJson(this);
    }
}

子类事件对象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import com.google.gson.Gson;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:35
 * @description: 子类事件对象
 */
public class AppleEvent extends FruitEvent {

    public AppleEvent() {
    }

    public AppleEvent(String name) {
        super(name);
    }

    @Override
    public String toString() {
        return new Gson().toJson(this);
    }

}

监听器对象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import com.google.common.eventbus.Subscribe;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:32
 * @description: 一个监听子类对象，一个监听父类对象
 */
public class SimpleListener {

    @Subscribe
    public void acceptFruitEvent(FruitEvent fruitEvent) {
        System.out.println("SimpleListener.acceptFruitEvent => fruitEvent " + fruitEvent);
    }

    @Subscribe
    public void acceptAppleEvent(AppleEvent appleEvent) {
        System.out.println("SimpleListener.acceptAppleEvent => appleEvent " + appleEvent);
    }
}

单元测试：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import com.google.common.eventbus.EventBus;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:32
 * @description: event 如果有父类，此时某个监听器监听的event是这个基类，
 * 那么消息是这个子类的时候，该监听父类的监听器方法也会执行
 */
public class MultipleEventExample {

    public static void main(String[] args) {
        EventBus eventBus = new EventBus();

        eventBus.register(new SimpleListener());

        eventBus.post(new AppleEvent("apple"));

        eventBus.post(new FruitEvent("fruit"));
    }
}

日志输出：

1
2
3
SimpleListener.acceptAppleEvent => appleEvent {"name":"apple"}
SimpleListener.acceptFruitEvent => fruitEvent {"name":"apple"}
SimpleListener.acceptFruitEvent => fruitEvent {"name":"fruit"}

异常处理

多个监听器同时监听相同消息，其中一个监听器有异常，那么不会影响其他监听器消费消息。可以在EventBus构造函数中传入一个异常处理器，当监听器有异常抛出时，会执行这个异常处理器。

思考点：如果异常处理很长时间，会不会导致后面的监听器等待执行？

异常处理器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import com.google.common.base.Throwables;
import com.google.common.eventbus.SubscriberExceptionContext;
import com.google.common.eventbus.SubscriberExceptionHandler;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:54
 * @description: 异常处理器
 */
public class MySubscriberExceptionHandler implements SubscriberExceptionHandler {

    @Override
    public void handleException(Throwable exception, SubscriberExceptionContext context) {
        System.out.println("======== handleException start =========");
        System.out.println("exception => " + Throwables.getStackTraceAsString(exception));
        System.out.println("EventBus => " + context.getEventBus());
        System.out.println("Subscriber => " + context.getSubscriber().getClass().getName());
        System.out.println("SubscriberMethod => " + context.getSubscriberMethod());
        System.out.println("======== handleException end =========");
    }
}

监听器对象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import com.google.common.eventbus.Subscribe;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:49
 * @description:
 */
public class MyListener {

    @Subscribe
    public void method1(Integer sum) {
        System.out.println("MyListener.method1 => " + sum);
    }

    @Subscribe
    public void method2(Integer sum) {
        System.out.println("MyListener.method2 => " + sum);
    }

    @Subscribe
    public void method3(Integer sum) {
        int i = sum / 0;
        System.out.println("MyListener.method3 => " + sum);
    }

}

单元测试：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import com.google.common.eventbus.EventBus;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 22:48
 * @description: 多个监听器同时监听相同消息，其中一个监听器有异常，那么不会影响其他监听器消费消息
 * 可以在EventBus构造函数中传入一个异常处理器，当监听器有异常抛出时，会执行这个异常处理器
 *
 * 思考点：如果异常处理很长时间，会不会导致后面的监听器等待执行
 *
 */
public class ExceptionListenerExample {

    public static void main(String[] args) {

        EventBus eventBus = new EventBus(new MySubscriberExceptionHandler());

        eventBus.register(new MyListener());

        eventBus.post(Integer.MAX_VALUE);
    }
}

日志输出：

1
2
3
4
5
6
7
8
9
10
MyListener.method2 => 2147483647
MyListener.method1 => 2147483647
======== handleException start =========
exception => java.lang.ArithmeticException: / by zero
	…… // 打印日志栈信息

EventBus => EventBus{default}
Subscriber => org.woodwhales.guava.eventbus.demo4.MyListener
SubscriberMethod => public void org.woodwhales.guava.eventbus.demo4.MyListener.method3(java.lang.Integer)
======== handleException end =========

DeadEvent 实现类型

监听器监听 DeadEvent 类型事件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import com.google.common.eventbus.DeadEvent;
import com.google.common.eventbus.Subscribe;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 23:05
 * @description:
 */
public class DeadEventListener {

    @Subscribe
    public void handle(DeadEvent deadEvent) {
        System.out.println("DeadEventListener.handle deadEvent => " + deadEvent);
        System.out.println("DeadEventListener.handle event => " + deadEvent.getEvent());
        System.out.println("DeadEventListener.handle source => " + deadEvent.getSource());
    }
}

单元测试：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import com.google.common.eventbus.EventBus;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.2 23:07
 * @description: 监听DeadEvent消息类型，那么可以从这个DeadEvent对象中获取事件源和eventBus
 */
public class DeadEventExample {

    public static void main(String[] args) {
        EventBus eventBus = new EventBus("DEAD-EVENT-BUS");

        eventBus.register(new DeadEventListener());

        eventBus.post("message");
    }
}

日志输出：

1
2
3
DeadEventListener.handle deadEvent => DeadEvent{source=EventBus{DEAD-EVENT-BUS}, event=message}
DeadEventListener.handle event => message
DeadEventListener.handle source => EventBus{DEAD-EVENT-BUS}

自实现 EventBus

Bus 接口设计

自己实现一个 EventBus 前需要先设计好需要多少功能，即多核心少接口：将监听者注册bus，将监听者从bus中移除，发送消息，发送带有主题的消息，获取当前 bus 名称等：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.4 13:55
 * @description: Bus 接口
 */
public interface MyBus {

    /**
     * 注册到 bus
     * @param subscriber 订阅者
     */
    void register(Object subscriber);

    /**
     * 从bus中取消注册
     * @param subscriber
     */
    void unRegister(Object subscriber);

    /**
     * 发送消息
     * @param event
     */
    void post(Object event);

    /**
     * 发送消息
     * @param event
     * @param topic
     */
    void post(Object event, String topic);


    /**
     * 关闭bus
     */
    void close();

    /**
     * 获取当前bus名称
     * @return
     */
    String getBusName();

}

Bus 接口实现

实现 Bus 接口，重要的是怎么来保存注册到中的监听者，怎么在消息发布者调用 post() 发布消息的时候，根据消息的类型找到对应的要监听的方法并执行，如果有异常处理器，则执行异常处理器。

MyEventBus 实现 MyBus 接口，在其类中增加以下属性：MyEventBus 的构造函数仅仅是设置属性的入口，消息存储和消息消费，均由 MyRegistry 和 MyDispatcher 来执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
public class MyEventBus implements MyBus {

    private final MyRegistry registry = new MyRegistry();

    private String busName;

    private static final String DEFAULT_BUS_NAME = "DEFAULT";

    private static final String DEFAULT_TOPIC = "default-topic";

    private final MyDispatcher dispatcher;
    
    public MyEventBus() {
        this(DEFAULT_BUS_NAME, null, MyDispatcher.SEQ_EXECUTOR_SERVICE);
    }

    public MyEventBus(MyEventExceptionHandler eventExceptionHandler) {
        this(DEFAULT_BUS_NAME, eventExceptionHandler, MyDispatcher.SEQ_EXECUTOR_SERVICE);
    }

    public MyEventBus(String busName, MyEventExceptionHandler eventExceptionHandler, Executor executor) {
        if(Objects.isNull(busName) || busName.isEmpty()) {
            busName = DEFAULT_BUS_NAME;
        }
        this.busName = busName;
        this.dispatcher = MyDispatcher.newDispatcher(executor, eventExceptionHandler);
    }
    
    @Override
    public void register(Object subscriber) {
        this.registry.bind(subscriber);
    }

    @Override
    public void unRegister(Object subscriber) {
        this.unRegister(subscriber);
    }

    @Override
    public void post(Object event) {
        this.post(event, DEFAULT_TOPIC);
    }

    @Override
    public void post(Object event, String topic) {
        this.dispatcher.dispatch(this, registry, event, topic);
    }

    @Override
    public void close() {
        this.dispatcher.close();
    }

    @Override
    public String getBusName() {
        return this.busName;
    }
}

Dispatcher 实现

Dispatcher 用于执行消息消费，因此对于 Dispatcher，它需要直到要消费的消息信息及异常处理器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import java.lang.reflect.Method;
import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.4 14:33
 * @description: 消息执行器
 */
public class MyDispatcher {

    private final Executor executorService;

    private final MyEventExceptionHandler eventExceptionHandler;

    public static final Executor SEQ_EXECUTOR_SERVICE = SeqExecutorService.INSTANCE;

    public static final Executor PER_THREAD_EXECUTOR_SERVICE = PerThreadExecutorService.INSTANCE;

    private MyDispatcher(Executor executorService, MyEventExceptionHandler eventExceptionHandler) {
        this.executorService = executorService;
        this.eventExceptionHandler = eventExceptionHandler;
    }

    public void dispatch(MyBus bus, MyRegistry registry, Object event, String topic) {
        ConcurrentLinkedQueue<MySubscriber> subscribers = registry.scanSubscriber(topic);
        if(Objects.isNull(subscribers)) {
            if(Objects.nonNull(eventExceptionHandler)) {
                eventExceptionHandler.handle(new IllegalArgumentException(String.format("The topic [%s] not bind yet", topic)),
                        new DefaultMyContext(bus.getBusName(), null, event));
            }

            return;
        }

        // 遍历所有的监听者，拿到对应的事件参数对象，比对消息的事件类型和监听者参数事件类型是否一致
        subscribers.stream().filter(subscriber -> !subscriber.isDisabled())
                .filter(subscriber -> {
                    Method subscribeMethod = subscriber.getSubscribeMethod();
                    Class<?> parameterTypeClass = subscribeMethod.getParameterTypes()[0];
                    return parameterTypeClass.isAssignableFrom(event.getClass());
                })
                .forEach(subscriber -> invokeRealSubscribe(bus, subscriber, event));
    }

    private void invokeRealSubscribe(MyBus bus, MySubscriber subscriber, Object event) {

        Method subscribeMethod = subscriber.getSubscribeMethod();
        Object subscribeObject = subscriber.getSubscribeObject();

        this.executorService.execute(() -> {
            try {
                subscribeMethod.invoke(subscribeObject, event);
            } catch (Exception e) {
                if(Objects.nonNull(this.eventExceptionHandler)) {
                    this.eventExceptionHandler.handle(e, new DefaultMyContext(bus.getBusName(), subscriber, event));
                }
            }
        });
    }


    public static MyDispatcher newDispatcher(Executor executorService, MyEventExceptionHandler eventExceptionHandler) {
        return new MyDispatcher(executorService, eventExceptionHandler);
    }

    public static MyDispatcher newSeqDispatcher(MyEventExceptionHandler eventExceptionHandler) {
        return new MyDispatcher(SEQ_EXECUTOR_SERVICE, eventExceptionHandler);
    }

    public static MyDispatcher newPerThreadDispatcher(MyEventExceptionHandler eventExceptionHandler) {
        return new MyDispatcher(PER_THREAD_EXECUTOR_SERVICE, eventExceptionHandler);
    }

    public void close() {
        if(executorService instanceof ExecutorService) {
            ((ExecutorService) executorService).shutdown();
        }
    }

    private static class SeqExecutorService implements Executor {

        private static final SeqExecutorService INSTANCE = new SeqExecutorService();

        @Override
        public void execute(Runnable command) {
            command.run();
        }
    }

    private static class PerThreadExecutorService implements Executor {

        private static final PerThreadExecutorService INSTANCE = new PerThreadExecutorService();

        @Override
        public void execute(Runnable command) {
            new Thread(command).start();
        }
    }

    public static class DefaultMyContext implements MyEventContext {

        private final String busName;

        private final MySubscriber subscriber;

        private final Object event;

        public DefaultMyContext(String busName, MySubscriber subscriber, Object event) {
            this.busName = busName;
            this.subscriber = subscriber;
            this.event = event;
        }

        @Override
        public String getSource() {
            return this.busName;
        }

        @Override
        public Object getSubscriberObject() {
            return Objects.nonNull(subscriber) ? subscriber.getSubscribeObject() : null;
        }

        @Override
        public Method getSubscribeMethod() {
            return Objects.nonNull(subscriber) ? subscriber.getSubscribeMethod() : null;
        }

        @Override
        public Object getEvent() {
            return this.event;
        }
    }
}

MyRegistry 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.4 14:04
 * @description: 已订阅bus的订阅者容器对象
 */
public class MyRegistry {

    private final ConcurrentHashMap<String, ConcurrentLinkedQueue<MySubscriber>> subscriberContainer = new ConcurrentHashMap<> ();

    /**
     * 绑定
     * @param subscriber
     */
    public void bind(Object subscriber) {
        List<Method> subscriberMethods = getSubscriberMethods(subscriber);
        subscriberMethods.forEach(method -> tireSubscriber(subscriber, method));
    }

    private void tireSubscriber(Object subscriber, Method method) {
        MySubscribe mySubscribe = method.getDeclaredAnnotation(MySubscribe.class);
        String topic = mySubscribe.topic();
        subscriberContainer.computeIfAbsent(topic, key -> new ConcurrentLinkedQueue<>());
        subscriberContainer.get(topic).add(new MySubscriber(subscriber, method));
    }

    private List<Method> getSubscriberMethods(Object subscriber) {
        List<Method> methods = new ArrayList<>();

        Class<?> subscriberClass = subscriber.getClass();
        while (Objects.nonNull(subscriberClass)) {
            Method[] declaredMethods = subscriberClass.getDeclaredMethods();

            Arrays.stream(declaredMethods).filter(method ->
                method.isAnnotationPresent(MySubscribe.class) &&
                        method.getParameterCount() == 1 &&
                        method.getModifiers() == Modifier.PUBLIC
            ).forEach(methods::add);

            // 获取当前 subscriber 的父类
            subscriberClass = subscriberClass.getSuperclass();
        }

        return methods;
    }

    /**
     * 解绑
     * @param subscriber
     */
    public void unBind(Object subscriber) {
        subscriberContainer.forEach((key, queue) -> {
            queue.forEach(subscriberItem -> {
                if(subscriberItem.getSubscribeObject() == subscriber) {
                    subscriberItem.setDisabled(false);
                }
            });
        });
    }

    /**
     * 根据topic获取已经绑定的监听对象
     * @param topic
     * @return
     */
    public ConcurrentLinkedQueue<MySubscriber> scanSubscriber(final String topic) {
        return subscriberContainer.get(topic);
    }

}

EventExceptionHandler 异常处理器

当有异常发生时，用户可以自己实现异常处理器接口，当有异常发生时会回调该接口：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.4 14:14
 * @description: 异常处理器
 */
public interface MyEventExceptionHandler {

    /**
     * 对 event 异常处理
     * @param cause
     * @param eventContext
     */
    void handle(Throwable cause, MyEventContext eventContext);

}

消息上下文对象

异常产生，需要告知消费者，产生异常的相关上下文：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import java.lang.reflect.Method;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.4 14:28
 * @description:
 */
public interface MyEventContext {

    String getSource();

    Object getSubscriberObject();

    Method getSubscribeMethod();

    Object getEvent();

}

Subscribe 监听器注解

监听器注解用于标记要监听并执行的方法：

1
2
3
4
5
6
7
8
9
10
11
12
/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.4 15:00
 * @description: 自定义监听器订阅方法注解
 */
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface MySubscribe {

    String topic() default "default-topic";
}

监听者信息对象

为了便于管理监听者对象，需要对监听者的信息进行封装，并存储到 MyRegistry 的 ConcurrentHashMap<> 容器中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import java.lang.reflect.Method;

/**
 * @projectName: guava
 * @author: woodwhales
 * @date: 20.7.4 15:05
 * @description: 监听者信息对象
 */
public class MySubscriber {

    private final Object subscribeObject;

    private final Method subscribeMethod;

    private boolean disabled = false;

    public MySubscriber(Object subscribeObject, Method subscribeMethod) {
        this.subscribeObject = subscribeObject;
        this.subscribeMethod = subscribeMethod;
    }

    public Object getSubscribeObject() {
        return subscribeObject;
    }

    public Method getSubscribeMethod() {
        return subscribeMethod;
    }

    public boolean isDisabled() {
        return disabled;
    }

    public void setDisabled(boolean disabled) {
        this.disabled = disabled;
    }
}

小结

自实现一个简单的 EventBus，要涉及到以下核心组件：

核心接口

MyBus

通过构造器接收监听者（使用 @MySubscribe 注解了某些方法）和自定义异常处理器（自实现了 MyEventExceptionHandler 接口）

MyEventExceptionHandler

用于消息过程出现异常时，处理异常

MyEventContext

异常处理过程中，将产生异常的数据封装起来

核心类

MyEventBus

MyBus 接口的实现类，其内部维护了 MyDispatcher 和 MyRegistry。

MyDispatcher

用于真正指定消费逻辑，内部维护了 MyEventExceptionHandler 实现（从 MyEventBus 构造器传入的）。

MyRegistry

用于对所有加了 @MySubscribe 注解的监听者信息进行管理，包括注册和注销。

MySubscriber

对注册的监听者信息进行封装，方便 MyRegistry 存储，也方便 MyDispatcher 消费执行。

核心注解

MySubscribe

提供要监听消息的方法注解标记。

updated updated 2024-01-05 2024-01-05
本文结束感谢阅读

本文标题:Guava EventBus 使用介绍及自实现详解

本文作者:

微信公号:木鲸鱼 | woodwhales

原始链接:https://woodwhales.cn/2020/07/06/072/

许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。

0%