您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 8 實現觀察者模式進階：線程安全、Mixin 封裝與異步通知

編輯：Java 相關知識

前言

承接上篇基本實現，本文深入探討生產環境級觀察者模式的關鍵議題：線程安全、代碼復用封裝、多方法監聽器適配、複雜/阻塞監聽器處理。

線程安全實現

問題場景

多線程並發註冊/移除監聽器
通知過程中監聽器列表被修改
通知順序與註冊順序一致性要求

方案一：方法級同步


public synchronized Listener registerListener(Listener l) { ... }
public synchronized void unregisterListener(Listener l) { ... }
public synchronized void notifyListeners(Event e) { ... }

缺點：鎖粒度過粗，讀操作（通知）互斥，吞吐率低。

方案二：讀寫鎖


public class ThreadSafeSubject {
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final Lock readLock = rwLock.readLock();
    private final Lock writeLock = rwLock.writeLock();
    private final List listeners = new ArrayList<>();

    public Listener register(Listener l) {
        writeLock.lock();
        try { listeners.add(l); return l; }
        finally { writeLock.unlock(); }
    }

    public void unregister(Listener l) {
        writeLock.lock();
        try { listeners.remove(l); }
        finally { writeLock.unlock(); }
    }

    public void notify(Event e) {
        readLock.lock();
        try { listeners.forEach(l -> l.onEvent(e)); }
        finally { readLock.unlock(); }
    }
}

優點：多讀併發，讀寫互斥，效能大幅提升。

方案三：公平鎖保證通知順序


private final ReadWriteLock rwLock = new ReentrantReadWriteLock(true); // 公平模式

確保線程按請求順序獲取鎖，通知順序與註冊/事件觸發順序一致。

監聽器內部線程安全

主體只負責列表訪問安全，監聽器邏輯安全由監聽器自行保證：


// 原子計數器
public class CountingListener implements Listener {
    private final AtomicLong count = new AtomicLong();
    public void onEvent(Event e) {
        System.out.println("計數: " + count.incrementAndGet());
    }
}

// 方法同步
public class SyncListener implements Listener {
    private int count = 0;
    public synchronized void onEvent(Event e) {
        System.out.println("計數: " + ++count);
    }
}

主體邏輯封裝為 Mixin

通用 ObservableMixin


public abstract class ObservableMixin {
    private final List listeners = new ArrayList<>();
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final Lock rLock = rwLock.readLock();
    private final Lock wLock = rwLock.writeLock();

    public L register(L listener) {
        wLock.lock();
        try { listeners.add(listener); return listener; }
        finally { wLock.unlock(); }
    }

    public void unregister(L listener) {
        wLock.lock();
        try { listeners.remove(listener); }
        finally { wLock.unlock(); }
    }

    // 函數式通知，支援任意消費邏輯
    public void notify(Consumer action) {
        rLock.lock();
        try { listeners.forEach(action); }
        finally { rLock.unlock(); }
    }
}

業務類繼承 Mixin


public class Zoo extends ObservableMixin {
    private final List animals = new ArrayList<>();

    public void addAnimal(Animal a) {
        animals.add(a);
        notify(l -> l.onAnimalAdded(a)); // Lambda 表達式傳遞具體通知邏輯
    }

    public void removeAnimal(Animal a) {
        animals.remove(a);
        notify(l -> l.onAnimalRemoved(a));
    }
}

多監聽器類型合併策略

Java 不支援多重繼承泛型類，解決方案：


// 合併為單一介面
public interface ZooListener {
    void onAnimalAdded(Animal a);
    void onAnimalRemoved(Animal a);
}

public class Zoo extends ObservableMixin {
    // 同前，通知時調用對應方法
}

多方法監聽器與適配器模式

問題

監聽器介面方法過多（如 Swing MouseListener 5 方法），實現類被迫實現所有方法。

適配器類


public class ZooAdapter implements ZooListener {
    public void onAnimalAdded(Animal a) {}
    public void onAnimalRemoved(Animal a) {}
}

具體監聽者僅覆寫關心方法


public class NamePrintingAdapter extends ZooAdapter {
    @Override
    public void onAnimalAdded(Animal a) {
        System.out.println("新增動物: " + a.getName());
    }
}

Java 8 default 方法替代適配器


public interface ZooListener {
    default void onAnimalAdded(Animal a) {}
    default void onAnimalRemoved(Animal a) {}
}

注意：default 方法旨在介面演進兼容，濫用可能導致介面膨脹，需謹慎權衡。

合併介面到具體類


public class ZooListener {
    public void onAnimalAdded(Animal a) {}
    public void onAnimalRemoved(Animal a) {}
}
// 直接繼承具體類，但失去多介面實現能力

複雜與阻塞監聽器處理

挑戰

監聽器執行耗時長（IO、計算、網路請求）
阻塞主線程/事件發布線程
需保證通知順序與異常隔離

策略一：監聽器內部啟動新線程


public void onEvent(Event e) {
    new Thread(() -> heavyProcessing(e)).start();
}

缺點：線程管理分散，難以控制併發度。

策略二：主體異步通知


public void notifyAsync(Event e) {
    executor.submit(() -> notify(e)); // 使用線程池
}

需配合：線程安全的監聽器列表訪問（複製快照或讀寫鎖）。

策略三：命令隊列 + 消費者線程池 (生產者-消費者)


public class EventFunctor {
    private final Listener listener;
    private final Event event;
    private final long timestamp; // 保留事件發生時間

    public EventFunctor(Listener l, Event e) {
        this.listener = l; this.event = e; this.timestamp = System.nanoTime();
    }
    public void execute() { listener.onEvent(event); }
}

// 發布端
BlockingQueue queue = new LinkedBlockingQueue<>();
public void notify(Event e) {
    listeners.forEach(l -> queue.offer(new EventFunctor(l, e)));
}

// 消費者線程
while (running) {
    EventFunctor f = queue.take();
    f.execute(); // 可在這裡添加超時、重試、熔斷
}

優點：解耦發布與執行、可控併發、支援優先級隊列、順序保證、背壓處理。

完整生產級範例


public class ProductionSubject {
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock(true);
    private final List listeners = new ArrayList<>();
    private final ExecutorService executor = Executors.newFixedThreadPool(4);
    private final BlockingQueue notificationQueue = new LinkedBlockingQueue<>();

    // 啟動消費者線程
    { startNotificationConsumers(2); }

    public L register(L l) {
        wLock(); try { listeners.add(l); return l; } finally { wUnlock(); }
    }
    public void unregister(L l) { wLock(); try { listeners.remove(l); } finally { wUnlock(); } }

    public void notify(Consumer action) {
        // 快照模式避免長時間持有讀鎖
        List snapshot;
        rLock(); try { snapshot = new ArrayList<>(listeners); } finally { rUnlock(); }
        
        // 封裝為任務提交隊列
        for (L listener : snapshot) {
            notificationQueue.offer(() -> {
                try { action.accept(listener); }
                catch (Exception ex) { logError(listener, ex); }
            });
        }
    }

    private void startNotificationConsumers(int count) {
        for (int i = 0; i < count; i++) {
            executor.submit(() -> {
                while (!Thread.interrupted()) {
                    try { notificationQueue.take().run(); }
                    catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
                    catch (Exception e) { logError("consumer", e); }
                }
            });
        }
    }
    // ... 輔助方法
}