基于DolphinScheduler抽取通用EventBus组件：支持耽误与事件驱动 ...

一、思路来源

虽然guava中的eventbus已经很方便了，但是还是想要实现一个更为方便，同时支持耽误事件、同时带eventbus的组件。在Apache DolphinScheduler项目中，有一个eventbus的组件，这个组件写得挺好的，想着用在业务系统上，因此本身抽取了一下，拿到业务系统中来用。话不多说，我们把它抽取出来吧，同时举行demo的运行。还是要感谢Apache DolphinScheduler的开源，让这个很简单，但是很高效的组件可以或许让我们便捷地使用。
二、具体实现过程

起首是定义事件接口：

1. public interface IEvent {
2. }

复制代码

针对事件接口，我们抽象出共性方法接口：耽误时间和过期时间。

1. public abstractclass AbstractDelayEvent implements IEvent, Delayed {
2.     privatefinallong delayTime;
3.     privatefinallong expireTime;
6.     public long getDelayTime() {
7.         return delayTime;
8.     }
10.     public long getExpireTime() {
11.         return expireTime;
12.     }
14.     public AbstractDelayEvent(long delayTime) {
15.         this.delayTime = delayTime;
16.         this.expireTime = System.currentTimeMillis() + delayTime;
17.     }
19.     @Override
20.     public long getDelay(TimeUnit unit) {
21.         long diff = expireTime - System.currentTimeMillis();
22.         return unit.convert(diff, TimeUnit.MILLISECONDS);
23.     }
25.     @Override
26.     public int compareTo(Delayed o) {
27.         if (this.expireTime < ((AbstractDelayEvent) o).expireTime) {
28.             return -1;
29.         }
30.         if (this.expireTime > ((AbstractDelayEvent) o).expireTime) {
31.             return1;
32.         }
33.         return0;
34.     }
35. }

复制代码

主要的信息：
定义eventbus中，我们需要使用的方法：

1. public interface IEventBus<T extends IEvent> {
3.     void publish(T event);
5.     Optional<T> poll() throws InterruptedException;
7.     Optional<T> peek();
9.     Optional<T> remove();
11.     boolean isEmpty();
13.     int size();
14. }

复制代码

可以看到主要是：发布事件、消费、移除、删除、判断当前的事件是否为空，以及事件大小等方法。其中最重要的方法为发布事件和消费处置惩罚事件方法。
针对当前的事件bus接口举行抽象，抽取出共性方法，方便复用：

1. public abstractclass AbstractDelayEventBus<T extends AbstractDelayEvent> implements IEventBus<T> {
3.     protectedfinal DelayQueue<T> delayEventQueue = new DelayQueue<>();
5.     @Override
6.     public void publish(T event) {
7.         delayEventQueue.put(event);
8.     }
10.     @Override
11.     public Optional<T> poll() throws InterruptedException {
12.         // 使用带超时的 poll 方法，等待事件到期
13.         return Optional.ofNullable(delayEventQueue.poll(1000, TimeUnit.MILLISECONDS));
14.     }
16.     @Override
17.     public Optional<T> peek() {
18.         return Optional.ofNullable(delayEventQueue.peek());
19.     }
21.     @Override
22.     public Optional<T> remove() {
23.         return Optional.ofNullable(delayEventQueue.poll());
24.     }
26.     @Override
27.     public boolean isEmpty() {
28.         return delayEventQueue.isEmpty();
29.     }
31.     @Override
32.     public int size() {
33.         return delayEventQueue.size();
34.     }
35. }

复制代码

三、测试运行效果

接下来，我们使用它，来举行处置惩罚：
定义本身的耽误事件：
如果是在业务中，可以定义本身的业务数据信息事件对象

1. public class MyDelayEvent extends AbstractDelayEvent {
2.     private final String message;
4.     public MyDelayEvent(long delayTime, String message) {
5.         super(delayTime);
6.         this.message = message;
7.     }
9.     public String getMessage() {
10.         return message;
11.     }
12. }

复制代码

定义事件耽误事件bus
当然也可以举行本身的可定制化特性。

1. public class MyDelayEventBus extends AbstractDelayEventBus<MyDelayEvent> {
2.     // 不需要额外的修改
3. }

复制代码

举行测试：
思路：创建事件总线、发布事件，然后针对发布的事件信息，举行消费，然后等待耽误时间的到来，从而实现消费，从而举行业务的处置惩罚。

1. import java.util.Optional;
3. publicclass EventBusExample {
4.     public static void main(String[] args) throws InterruptedException {
5.         // 创建事件总线
6.         IEventBus<MyDelayEvent> eventBus = new MyDelayEventBus();
8.         // 发布单个事件
9.         eventBus.publish(new MyDelayEvent(100, "Single Event"));
10.         System.out.println("After publish, event bus size: " + eventBus.size());
12.         // 持续尝试消费事件
13.         while (true) {
14.             Optional<MyDelayEvent> event = eventBus.poll();
15.             if (event.isPresent()) {
16.                 System.out.println("Received event: " + event.get().getMessage());
17.             } else {
18.                 System.out.println("No event received within the timeout.");
19.                 break;
20.             }
21.         }
23.         // 检查总线大小
24.         System.out.println("Event bus size: " + eventBus.size());
25.     }
26. }

复制代码

运行结果：

可以看到实现本身的业务逻辑还是很方便的，可以本身实现吧，这里给出的代码是可以运行的。

• 源码地点：https://gitee.com/null_713_2407/pratice
  
• 参考：
  github：https://github.com/apache/dolphinscheduler
  

作者 | 刘亚洲

本文由 白鲸开源 提供发布支持！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。