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

Java 觀察者模式泛型實現：類型安全的事件通知機制

編輯：Java 相關知識

泛型觀察者模式核心優勢

傳統 java.util.Observable/Observer 使用 Object 傳遞資料，需強制轉型，失去編譯期型別安全。泛型版本在**編譯期保證通知資料型別正確**，消除 ClassCastException 風險。

被觀察者

public class Observable {
    private final List> observers = new ArrayList<>();
    private boolean changed = false;

    public synchronized void addObserver(Observer observer) {
        if (observer == null) throw new NullPointerException("observer == null");
        if (!observers.contains(observer)) observers.add(observer);
    }

    public synchronized void deleteObserver(Observer observer) {
        observers.remove(observer);
    }

    public synchronized void deleteObservers() {
        observers.clear();
    }

    public synchronized int countObservers() {
        return observers.size();
    }

    protected synchronized void setChanged() {
        changed = true;
    }

    protected synchronized void clearChanged() {
        changed = false;
    }

    public synchronized boolean hasChanged() {
        return changed;
    }

    // 無參數通知
    public void notifyObservers() {
        notifyObservers(null);
    }

    // 攜帶泛型資料通知
    public void notifyObservers(T data) {
        Observer[] observersSnapshot;
        synchronized (this) {
            if (!changed) return;
            clearChanged();
            observersSnapshot = observers.toArray(new Observer[0]);
        }
        // 釋放鎖後通知，避免觀察者回調中修改列表導致死鎖
        for (Observer observer : observersSnapshot) {
            observer.update(this, data);
        }
    }
}

關鍵設計點

快照模式：toArray 複製陣列，通知時釋放鎖，防死鎖
狀態標記：changed 避免重複通知，setChanged() 由具體業務邏輯調用
執行緒安全：列表修改與快照建立同步，通知過程無鎖
泛型傳遞：Observer 確保 update 參數型別一致

觀察者介面

public interface Observer {
    void update(Observable observable, T data);
}

使用範例：股價監控系統

領域模型

public class StockPrice {
    private final String symbol;
    private final double price;
    private final long timestamp;
    // 建構子、getter、toString
}

具體被觀察者

public class StockTicker extends Observable {
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private final Map basePrices = new HashMap<>();

    public StockTicker() {
        basePrices.put("AAPL", 150.0);
        basePrices.put("GOOGL", 2800.0);
        basePrices.put("TSLA", 800.0);
    }

    public void start() {
        scheduler.scheduleAtFixedRate(this::simulatePriceChange, 0, 1, TimeUnit.SECONDS);
    }

    private void simulatePriceChange() {
        String symbol = getRandomSymbol();
        double base = basePrices.get(symbol);
        double change = (Math.random() - 0.5) * 10;  // ±5%
        double newPrice = Math.max(0.01, base + change);
        basePrices.put(symbol, newPrice);
        
        StockPrice price = new StockPrice(symbol, newPrice, System.currentTimeMillis());
        setChanged();      // 標記有變更
        notifyObservers(price);  // 通知所有觀察者
    }

    private String getRandomSymbol() {
        List symbols = new ArrayList<>(basePrices.keySet());
        return symbols.get((int)(Math.random() * symbols.size()));
    }

    public void stop() {
        scheduler.shutdown();
    }
}

具體觀察者實作

// 控制台記錄觀察者
public class ConsoleLogger implements Observer {
    @Override
    public void update(Observable observable, StockPrice data) {
        System.out.printf("[LOG] %s: %.2f @ %d%n", data.getSymbol(), data.getPrice(), data.getTimestamp());
    }
}

// 告警觀察者：價格變動超過閾值觸發
public class PriceAlert implements Observer {
    private final double thresholdPercent;
    private final Map lastPrices = new HashMap<>();

    public PriceAlert(double thresholdPercent) {
        this.thresholdPercent = thresholdPercent;
    }

    @Override
    public void update(Observable observable, StockPrice data) {
        String symbol = data.getSymbol();
        double current = data.getPrice();
        Double last = lastPrices.get(symbol);
        
        if (last != null) {
            double changePercent = Math.abs((current - last) / last) * 100;
            if (changePercent >= thresholdPercent) {
                System.out.printf("🚨 ALERT: %s 變動 %.2f%% (%.2f → %.2f)%n", 
                    symbol, changePercent, last, current);
            }
        }
        lastPrices.put(symbol, current);
    }
}

// 統計觀察者：維護移動平均
public class MovingAverageTracker implements Observer {
    private final int windowSize;
    private final Map> windows = new HashMap<>();

    public MovingAverageTracker(int windowSize) {
        this.windowSize = windowSize;
    }

    @Override
    public void update(Observable observable, StockPrice data) {
        String symbol = data.getSymbol();
        windows.computeIfAbsent(symbol, k -> new ArrayDeque<>()).addLast(data.getPrice());
        Deque window = windows.get(symbol);
        if (window.size() > windowSize) window.removeFirst();
        
        double avg = window.stream().mapToDouble(Double::doubleValue).average().orElse(0);
        System.out.printf("[AVG-%d] %s: 當前=%.2f, 平均=%.2f%n", windowSize, symbol, data.getPrice(), avg);
    }
}

客戶端整合

public class ObserverDemo {
    public static void main(String[] args) throws InterruptedException {
        StockTicker ticker = new StockTicker();
        
        // 註冊多型觀察者
        ticker.addObserver(new ConsoleLogger());
        ticker.addObserver(new PriceAlert(3.0));      // 3% 告警
        ticker.addObserver(new MovingAverageTracker(5)); // 5筆移動平均
        
        ticker.start();
        Thread.sleep(10000);  // 運行 10 秒
        ticker.stop();
    }
}

進階擴展：非同步通知與錯誤隔離

public class AsyncObservable extends Observable {
    private final ExecutorService executor;

    public AsyncObservable(ExecutorService executor) {
        this.executor = executor;
    }

    @Override
    public void notifyObservers(T data) {
        Observer[] snapshot;
        synchronized (this) {
            if (!hasChanged()) return;
            clearChanged();
            snapshot = observers.toArray(new Observer[0]);
        }
        
        // 非同步並行通知，單一觀察者異常不影響其他
        for (Observer observer : snapshot) {
            executor.submit(() -> {
                try {
                    observer.update(this, data);
                } catch (Exception e) {
                    System.err.println("Observer error: " + e.getMessage());
                }
            });
        }
    }
}

與 JDK 內建對比

特性	java.util.Observable	泛型版本
資料型別	`Object` (需轉型)	`T` (編譯期安全)
執行緒安全	同步方法	快照模式 + 無鎖通知
非同步支援	無	可繼承擴展
錯誤隔離	無 (一個觀察者拋異常中斷後續)	可在擴展中實現
類別繼承	類 (單繼承限制)	類 (同樣限制，建議組合)

最佳實踐建議

優先組合而非繼承：將 Observable 作為欄位內嵌，避免單繼承限制
觀察者保持輕量：update 中避免耗時操作，必要時自行異步處理
生命週期管理：及時 deleteObserver 防止記憶體洩漏
不可變資料物件：通知傳遞 final 類別，防止觀察者修改共享狀態
考慮 Reactor/RxJava：複雜事件流場景下，響應式流庫提供更豐富操作子

總結

泛型觀察者模式以極小代價（幾行泛型宣告）換取**編譯期型別安全**、**執行緒安全通知**、**錯誤隔離擴展性**。這是 Java 中實現鬆耦合事件驅動架構的基石模式，適用於 GUI 事件、訊息總線、快取失效通知、監控告警等眾多場景。掌握其核心實現細節（快照、狀態標記、泛型傳遞），能避開併發陷阱，構建健壯可維護的系統。