1. Background and Motivation Although Google Guava’s EventBus is already quite convenient, I wanted to build something more extensible — a component that not only functions as an EventBus but also supports delayed events out of the box. EventBus delayed events While exploring the Apache DolphinScheduler project, I found its built-in eventbus component to be a well-written and efficient solution. It inspired me to extract and adapt it for use in our own business systems. Apache DolphinScheduler eventbus Thanks to DolphinScheduler’s open-source nature, this component can now be conveniently reused with minimal overhead. Let’s walk through how to extract it and run a working demo. 2. Implementation Details Step 1: Define the Event Interface First, we define a base interface for all events: public interface IEvent {
} public interface IEvent {
} Step 2: Abstract Class for Delayed Events To support delayed execution, we introduce an abstract class AbstractDelayEvent that extends Delayed and implements IEvent. It includes both the delay time and the expiration time. AbstractDelayEvent Delayed IEvent delay time expiration time public abstract class AbstractDelayEvent implements IEvent, Delayed {
    private final long delayTime;
    private final long expireTime;

    public long getDelayTime() {
        return delayTime;
    }

    public long getExpireTime() {
        return expireTime;
    }

    public AbstractDelayEvent(long delayTime) {
        this.delayTime = delayTime;
        this.expireTime = System.currentTimeMillis() + delayTime;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long diff = expireTime - System.currentTimeMillis();
        return unit.convert(diff, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        if (this.expireTime < ((AbstractDelayEvent) o).expireTime) {
            return -1;
        }
        if (this.expireTime > ((AbstractDelayEvent) o).expireTime) {
            return 1;
        }
        return 0;
    }
} public abstract class AbstractDelayEvent implements IEvent, Delayed {
    private final long delayTime;
    private final long expireTime;

    public long getDelayTime() {
        return delayTime;
    }

    public long getExpireTime() {
        return expireTime;
    }

    public AbstractDelayEvent(long delayTime) {
        this.delayTime = delayTime;
        this.expireTime = System.currentTimeMillis() + delayTime;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long diff = expireTime - System.currentTimeMillis();
        return unit.convert(diff, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        if (this.expireTime < ((AbstractDelayEvent) o).expireTime) {
            return -1;
        }
        if (this.expireTime > ((AbstractDelayEvent) o).expireTime) {
            return 1;
        }
        return 0;
    }
} Step 3: Define the EventBus Interface Next, we define the interface for the EventBus, with core operations like publish, poll, peek, remove, and state checks: public interface IEventBus<T extends IEvent> {

    void publish(T event);

    Optional<T> poll() throws InterruptedException;

    Optional<T> peek();

    Optional<T> remove();

    boolean isEmpty();

    int size();
} public interface IEventBus<T extends IEvent> {

    void publish(T event);

    Optional<T> poll() throws InterruptedException;

    Optional<T> peek();

    Optional<T> remove();

    boolean isEmpty();

    int size();
} Step 4: Abstract Delay EventBus Implementation We then provide an abstract base class AbstractDelayEventBus, which implements the core logic using Java’s DelayQueue. AbstractDelayEventBus DelayQueue public abstract class AbstractDelayEventBus<T extends AbstractDelayEvent> implements IEventBus<T> {

    protected final DelayQueue<T> delayEventQueue = new DelayQueue<>();

    @Override
    public void publish(T event) {
        delayEventQueue.put(event);
    }

    @Override
    public Optional<T> poll() throws InterruptedException {
        return Optional.ofNullable(delayEventQueue.poll(1000, TimeUnit.MILLISECONDS));
    }

    @Override
    public Optional<T> peek() {
        return Optional.ofNullable(delayEventQueue.peek());
    }

    @Override
    public Optional<T> remove() {
        return Optional.ofNullable(delayEventQueue.poll());
    }

    @Override
    public boolean isEmpty() {
        return delayEventQueue.isEmpty();
    }

    @Override
    public int size() {
        return delayEventQueue.size();
    }
} public abstract class AbstractDelayEventBus<T extends AbstractDelayEvent> implements IEventBus<T> {

    protected final DelayQueue<T> delayEventQueue = new DelayQueue<>();

    @Override
    public void publish(T event) {
        delayEventQueue.put(event);
    }

    @Override
    public Optional<T> poll() throws InterruptedException {
        return Optional.ofNullable(delayEventQueue.poll(1000, TimeUnit.MILLISECONDS));
    }

    @Override
    public Optional<T> peek() {
        return Optional.ofNullable(delayEventQueue.peek());
    }

    @Override
    public Optional<T> remove() {
        return Optional.ofNullable(delayEventQueue.poll());
    }

    @Override
    public boolean isEmpty() {
        return delayEventQueue.isEmpty();
    }

    @Override
    public int size() {
        return delayEventQueue.size();
    }
} 3. Demo: Testing the Component Let’s build a small example to see it in action. Define a Custom Delayed Event In a real system, this would hold your domain-specific business data. public class MyDelayEvent extends AbstractDelayEvent {
    private final String message;

    public MyDelayEvent(long delayTime, String message) {
        super(delayTime);
        this.message = message;
    }

    public String getMessage() {
        return message;
    }
} public class MyDelayEvent extends AbstractDelayEvent {
    private final String message;

    public MyDelayEvent(long delayTime, String message) {
        super(delayTime);
        this.message = message;
    }

    public String getMessage() {
        return message;
    }
} Implement Your Own EventBus (Optional) You can customize it further if needed. For simple use cases, this works as-is. public class MyDelayEventBus extends AbstractDelayEventBus<MyDelayEvent> {
    // No additional customization needed
} public class MyDelayEventBus extends AbstractDelayEventBus<MyDelayEvent> {
    // No additional customization needed
} Example: Publishing and Consuming Events Now, let’s run a simple main function that publishes and consumes delayed events: import java.util.Optional;

public class EventBusExample {
    public static void main(String[] args) throws InterruptedException {
        // Create event bus
        IEventBus<MyDelayEvent> eventBus = new MyDelayEventBus();

        // Publish a single event with 100ms delay
        eventBus.publish(new MyDelayEvent(100, "Single Event"));
        System.out.println("After publish, event bus size: " + eventBus.size());

        // Continuously try to consume events
        while (true) {
            Optional<MyDelayEvent> event = eventBus.poll();
            if (event.isPresent()) {
                System.out.println("Received event: " + event.get().getMessage());
            } else {
                System.out.println("No event received within the timeout.");
                break;
            }
        }

        // Final bus size
        System.out.println("Event bus size: " + eventBus.size());
    }
} import java.util.Optional;

public class EventBusExample {
    public static void main(String[] args) throws InterruptedException {
        // Create event bus
        IEventBus<MyDelayEvent> eventBus = new MyDelayEventBus();

        // Publish a single event with 100ms delay
        eventBus.publish(new MyDelayEvent(100, "Single Event"));
        System.out.println("After publish, event bus size: " + eventBus.size());

        // Continuously try to consume events
        while (true) {
            Optional<MyDelayEvent> event = eventBus.poll();
            if (event.isPresent()) {
                System.out.println("Received event: " + event.get().getMessage());
            } else {
                System.out.println("No event received within the timeout.");
                break;
            }
        }

        // Final bus size
        System.out.println("Event bus size: " + eventBus.size());
    }
} Output As you can see, this approach makes it easy to create and manage your own delay-capable EventBus system, enabling flexible event-driven programming for your business applications. 4. Source Code & References 🔗 Source Code: Gitee Repository
📚 Reference: Apache DolphinScheduler GitHub 🔗 Source Code: Gitee Repository Source Code: Gitee Repository 📚 Reference: Apache DolphinScheduler GitHub Reference: Apache DolphinScheduler GitHub If you're building delay-sensitive or event-driven systems, this lightweight and extendable component might be just what you need — and it’s inspired by one of the best open-source workflow schedulers out there.

Apache

Cybersecurity Giant Supercharges Apache SeaTunnel to Tame Complex Data 

Build Scalable Data Pipelines With Apache SeaTunnel × Cloudberry Integration

Read My Stories

Too Long; Didn't Read

Make resilience your competitive advantage

Build Your Own Delayed EventBus in Java with Just a Few Lines of Code

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

Untitled Story

A 10-Minute Deep Dive Into the Core Architecture of Apache SeaTunnel and DataX

How to Seamlessly Transfer From Airflow to Apache DolphinScheduler With Air2phin

A Comprehensive Guide to Building DolphinScheduler 3.2.0 Production-Grade Cluster Deployment

How to Regularly Delete Log Instances in DolphinScheduler

How to Enable Auto-Start for Apache DolphinScheduler

A 10-Minute Deep Dive Into the Core Architecture of Apache SeaTunnel and DataX

How to Seamlessly Transfer From Airflow to Apache DolphinScheduler With Air2phin

A Comprehensive Guide to Building DolphinScheduler 3.2.0 Production-Grade Cluster Deployment

How to Regularly Delete Log Instances in DolphinScheduler

How to Enable Auto-Start for Apache DolphinScheduler

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps