实现带优先级订阅者及并发修改支持的事件总线技术问询

Hey there! Let's break down how to build a type-safe event bus that meets all your requirements—handling subtype events, supporting priority-based subscriptions, and keeping everything type-checked at compile time. Here's a practical, production-ready approach:

Just to make sure we're aligned, you need:

• Type-safe subscriptions: No casting hacks, compile-time checks for event-subscriber compatibility
• Subtype event support: Subscribers registered for a parent event type should receive all its child events
• Priority control: Allow custom integer priorities for subscribers, plus a sensible default

1. Type-Safe Subscriber Interface #

Start with a generic Subscriber interface to enforce type safety. This ensures subscribers only handle events of the type they're designed for:

public interface Subscriber<T> {
    void onEvent(T event);
}

2. Subscription Wrapper with Priority #

Create a simple wrapper class to pair subscribers with their priority. This makes sorting and managing subscriptions straightforward:

public class Subscription<T> {
    private final Subscriber<T> subscriber;
    private final int priority;

    public Subscription(Subscriber<T> subscriber, int priority) {
        this.subscriber = subscriber;
        this.priority = priority;
    }

    // Getters for access in the event bus
    public Subscriber<T> getSubscriber() { return subscriber; }
    public int getPriority() { return priority; }
}

3. Core Event Bus Implementation #

This is the heart of the system. It manages subscriptions, handles subtype event routing, and enforces priority order:

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

public class EventBus {
    // Map event types to sorted lists of subscriptions (thread-safe)
    private final Map<Class<?>, List<Subscription<?>>> eventSubscriptions = new ConcurrentHashMap<>();
    // Default priority for subscribers that don't specify one
    public static final int DEFAULT_PRIORITY = 0;

    // Overload: Subscribe with default priority
    public <T> void subscribe(Class<T> eventType, Subscriber<T> subscriber) {
        subscribe(eventType, subscriber, DEFAULT_PRIORITY);
    }

    // Primary subscribe method: Register with custom priority
    public <T> void subscribe(Class<T> eventType, Subscriber<T> subscriber, int priority) {
        // Initialize list if it doesn't exist, then add the new subscription
        eventSubscriptions.computeIfAbsent(eventType, k -> new CopyOnWriteArrayList<>())
                .add(new Subscription<>(subscriber, priority));
        
        // Sort subscriptions by priority (higher values run first)
        eventSubscriptions.get(eventType).sort((s1, s2) -> Integer.compare(s2.getPriority(), s1.getPriority()));
    }

    // Unsubscribe a subscriber from a specific event type
    public <T> void unsubscribe(Class<T> eventType, Subscriber<T> subscriber) {
        List<Subscription<?>> subscriptions = eventSubscriptions.get(eventType);
        if (subscriptions != null) {
            subscriptions.removeIf(sub -> sub.getSubscriber().equals(subscriber));
            // Clean up empty lists to save memory
            if (subscriptions.isEmpty()) {
                eventSubscriptions.remove(eventType);
            }
        }
    }

    // Publish an event to all relevant subscribers (including parent types)
    @SuppressWarnings("unchecked")
    public <T> void publish(T event) {
        Class<?> eventClass = event.getClass();
        // Collect all parent types (superclasses + interfaces) of the event
        Set<Class<?>> relevantEventTypes = new HashSet<>();
        collectRelevantEventTypes(eventClass, relevantEventTypes);

        // Notify subscribers for each relevant type
        for (Class<?> type : relevantEventTypes) {
            List<Subscription<?>> subscriptions = eventSubscriptions.get(type);
            if (subscriptions != null) {
                for (Subscription<?> sub : subscriptions) {
                    // Safe cast: We registered the subscriber for this type, so it's compatible
                    ((Subscriber<T>) sub.getSubscriber()).onEvent(event);
                }
            }
        }
    }

    // Recursively collect all parent types of an event class
    private void collectRelevantEventTypes(Class<?> clazz, Set<Class<?>> types) {
        if (clazz == null || types.contains(clazz)) return;
        
        types.add(clazz);
        // Add all implemented interfaces
        for (Class<?> iface : clazz.getInterfaces()) {
            collectRelevantEventTypes(iface, types);
        }
        // Add superclass (walk up the inheritance tree)
        collectRelevantEventTypes(clazz.getSuperclass(), types);
    }
}

4. Usage Example #

Let's put this into action with a simple event hierarchy and subscribers:

// Define your event hierarchy
abstract class BaseEvent {}
class UserCreatedEvent extends BaseEvent {
    private final String username;
    public UserCreatedEvent(String username) { this.username = username; }
    public String getUsername() { return username; }
}
class UserDeletedEvent extends BaseEvent {
    private final String username;
    public UserDeletedEvent(String username) { this.username = username; }
    public String getUsername() { return username; }
}

// Create subscribers
class BaseEventHandler implements Subscriber<BaseEvent> {
    @Override
    public void onEvent(BaseEvent event) {
        System.out.println("Base handler got event: " + event.getClass().getSimpleName());
    }
}

class UserCreatedHandler implements Subscriber<UserCreatedEvent> {
    @Override
    public void onEvent(UserCreatedEvent event) {
        System.out.println("User created: " + event.getUsername() + " (high priority)");
    }
}

// Test the event bus
public class EventBusDemo {
    public static void main(String[] args) {
        EventBus bus = new EventBus();

        // Subscribe base handler with default priority
        bus.subscribe(BaseEvent.class, new BaseEventHandler());
        // Subscribe user created handler with higher priority (10 > 0)
        bus.subscribe(UserCreatedEvent.class, new UserCreatedHandler(), 10);

        // Publish a UserCreatedEvent
        bus.publish(new UserCreatedEvent("john_doe"));
        // Output order (priority matters):
        // User created: john_doe (high priority)
        // Base handler got event: UserCreatedEvent

        // Publish a UserDeletedEvent
        bus.publish(new UserDeletedEvent("jane_doe"));
        // Output:
        // Base handler got event: UserDeletedEvent
    }
}

• Type Safety: The generic Subscriber<T> and subscribe methods ensure you can't register a subscriber for an event type it can't handle—errors are caught at compile time, not runtime.
• Subtype Support: The collectRelevantEventTypes method recursively gathers all parent classes and interfaces of the published event, so subscribers registered for BaseEvent get notified of all its child events.
• Priority Control: Subscriptions are sorted in descending order of priority, so higher values run first. The DEFAULT_PRIORITY (0) gives a sensible fallback for quick subscriptions.
• Thread Safety: We use ConcurrentHashMap and CopyOnWriteArrayList to make the event bus safe for multi-threaded environments—critical for modular applications with concurrent module interactions.

内容的提问来源于stack exchange，提问作者Iavra