Go和Java實(shí)現(xiàn)觀察者模式

在監(jiān)控系統(tǒng)中，我們需要采集監(jiān)控指標(biāo)的信息，假設(shè)當(dāng)采集的指標(biāo)信息超過閾值時(shí)我們需要對該監(jiān)控指標(biāo)持久化到

數(shù)據(jù)庫中并且進(jìn)行告警。

本文通過指標(biāo)采集持久化和告警來說明觀察者模式的使用，使用Go語言和Java語言實(shí)現(xiàn)。

1、觀察者模式

觀察者模式是一種行為型設(shè)計(jì)模式，它定義了一種一對多的依賴關(guān)系，當(dāng)一個(gè)對象的狀態(tài)發(fā)生改變時(shí)，其所有依賴

者都會收到通知并自動更新。

當(dāng)對象間存在一對多關(guān)系時(shí)，則使用觀察者模式。比如，當(dāng)一個(gè)對象被修改時(shí)，則會自動通知依賴它的對象。觀察

者模式屬于行為型模式。

• 意圖：定義對象間的一種一對多的依賴關(guān)系，當(dāng)一個(gè)對象的狀態(tài)發(fā)生改變時(shí)，所有依賴于它的對象都得到通知

• 并被自動更新。

• 主要解決：一個(gè)對象狀態(tài)改變給其他對象通知的問題，而且要考慮到易用和低耦合，保證高度的協(xié)作。

• 何時(shí)使用：一個(gè)對象（目標(biāo)對象）的狀態(tài)發(fā)生改變，所有的依賴對象（觀察者對象）都將得到通知，進(jìn)行廣播

  通知。

• 如何解決：使用面向?qū)ο蠹夹g(shù)，可以將這種依賴關(guān)系弱化。

• 關(guān)鍵代碼：在抽象類里有一個(gè) ArrayList 存放觀察者們。

• 應(yīng)用實(shí)例：1、拍賣的時(shí)候，拍賣師觀察最高標(biāo)價(jià)，然后通知給其他競價(jià)者競價(jià)。 2、西游記里面悟空請求菩

  薩降服紅孩兒，菩薩灑了一地水招來一個(gè)老烏龜，這個(gè)烏龜就是觀察者，他觀察菩薩灑水這個(gè)動作。

• 優(yōu)點(diǎn)：1、觀察者和被觀察者是抽象耦合的。 2、建立一套觸發(fā)機(jī)制。

• 缺點(diǎn)：1、如果一個(gè)被觀察者對象有很多的直接和間接的觀察者的話，將所有的觀察者都通知到會花費(fèi)很多時(shí)

  間。 2、如果在觀察者和觀察目標(biāo)之間有循環(huán)依賴的話，觀察目標(biāo)會觸發(fā)它們之間進(jìn)行循環(huán)調(diào)用，可能導(dǎo)致系

  統(tǒng)崩潰。 3、觀察者模式?jīng)]有相應(yīng)的機(jī)制讓觀察者知道所觀察的目標(biāo)對象是怎么發(fā)生變化的，而僅僅只是知道

  觀察目標(biāo)發(fā)生了變化。

• 使用場景：

  一個(gè)抽象模型有兩個(gè)方面，其中一個(gè)方面依賴于另一個(gè)方面。將這些方面封裝在獨(dú)立的對象中使它們可以各自

  獨(dú)立地改變和復(fù)用。

  一個(gè)對象的改變將導(dǎo)致其他一個(gè)或多個(gè)對象也發(fā)生改變，而不知道具體有多少對象將發(fā)生改變，可以降低對象

  之間的耦合度。

  一個(gè)對象必須通知其他對象，而并不知道這些對象是誰。

  需要在系統(tǒng)中創(chuàng)建一個(gè)觸發(fā)鏈，A對象的行為將影響B(tài)對象，B對象的行為將影響C對象……，可以使用觀察者

  模式創(chuàng)建一種鏈?zhǔn)接|發(fā)機(jī)制。

• 注意事項(xiàng)：1、JAVA 中已經(jīng)有了對觀察者模式的支持類。 2、避免循環(huán)引用。 3、如果順序執(zhí)行，某一觀察者

  錯(cuò)誤會導(dǎo)致系統(tǒng)卡殼，一般采用異步方式。

• 觀察者模式包含以下幾個(gè)核心角色：

  主題（Subject）：也稱為被觀察者或可觀察者，它是具有狀態(tài)的對象，并維護(hù)著一個(gè)觀察者列表。主題提供

  了添加、刪除和通知觀察者的方法。

  觀察者（Observer）：觀察者是接收主題通知的對象。觀察者需要實(shí)現(xiàn)一個(gè)更新方法，當(dāng)收到主題的通知

  時(shí)，調(diào)用該方法進(jìn)行更新操作。

  具體主題（Concrete Subject）：具體主題是主題的具體實(shí)現(xiàn)類。它維護(hù)著觀察者列表，并在狀態(tài)發(fā)生改變時(shí)

  通知觀察者。

  具體觀察者（Concrete Observer）：具體觀察者是觀察者的具體實(shí)現(xiàn)類。它實(shí)現(xiàn)了更新方法，定義了在收到

  主題通知時(shí)需要執(zhí)行的具體操作。

觀察者模式通過將主題和觀察者解耦，實(shí)現(xiàn)了對象之間的松耦合。當(dāng)主題的狀態(tài)發(fā)生改變時(shí)，所有依賴于它的觀察

者都會收到通知并進(jìn)行相應(yīng)的更新。

2、Go實(shí)現(xiàn)觀察者模式

package observer

// ISubject通知者
type ISubject interface {
	Register(observer IObserver)
	Remove(observer IObserver)
	Notify(metric Metric)
}

package observer

// MetricSubject指標(biāo)通知者
type MetricSubject struct {
	observers []IObserver
}

type Metric struct {
	Name  string
	Value float64
	Time  string
}

func (metricSubject *MetricSubject) Register(observer IObserver) {
	metricSubject.observers = append(metricSubject.observers, observer)
}

func (metricSubject *MetricSubject) Remove(observer IObserver) {
	for i, ob := range metricSubject.observers {
		if ob == observer {
			metricSubject.observers = append(metricSubject.observers[:i], metricSubject.observers[i+1:]...)
		}
	}
}

func (metricSubject *MetricSubject) Notify(metric Metric) {
	for _, o := range metricSubject.observers {
		o.Update(metric)
	}
}

package observer

// IObserver觀察者
type IObserver interface {
	Update(metric Metric)
}

package observer

import "fmt"

// alert觀察者
type AlertObserver struct {
}

func (alertObserver * AlertObserver) Update(metric Metric){
   fmt.Printf("[%s] 指標(biāo) [%s] 的值為 [%.1f] 超過閾值,進(jìn)行告警!\n", metric.Time, metric.Name, metric.Value)
}

package observer

import "fmt"

// db觀察者
type DbObserver struct {
}

func (dbObserver *DbObserver) Update(metric Metric) {
	fmt.Printf("[%s] 指標(biāo) [%s] 的值為 [%.1f] 超過閾值,保存到數(shù)據(jù)庫!\n", metric.Time, metric.Name, metric.Value)
}

package main

import (
	"fmt"
	. "proj/observer"
	"time"
)

func main() {
	sub := &MetricSubject{}
	alert := &AlertObserver{}
	db := &DbObserver{}
	sub.Register(alert)
	sub.Register(db)
	sub.Notify(Metric{Name: "CPU", Value: 90, Time: time.Now().Format("2006-01-02 15:04:05")})
	sub.Notify(Metric{Name: "Memory", Value: 80, Time: time.Now().Format("2006-01-02 15:04:05")})
	fmt.Println("==========")
	// 后面不需要進(jìn)行持久化了
	sub.Remove(db)
	sub.Notify(Metric{Name: "CPU", Value: 90, Time: time.Now().Format("2006-01-02 15:04:05")})
	sub.Notify(Metric{Name: "Memory", Value: 80, Time: time.Now().Format("2006-01-02 15:04:05")})
}

# 程序輸出
[2023-07-11 22:05:34] 指標(biāo) [CPU] 的值為 [90.0] 超過閾值,進(jìn)行告警!
[2023-07-11 22:05:34] 指標(biāo) [CPU] 的值為 [90.0] 超過閾值,保存到數(shù)據(jù)庫!
[2023-07-11 22:05:34] 指標(biāo) [Memory] 的值為 [80.0] 超過閾值,進(jìn)行告警!
[2023-07-11 22:05:34] 指標(biāo) [Memory] 的值為 [80.0] 超過閾值,保存到數(shù)據(jù)庫!
==========
[2023-07-11 22:05:34] 指標(biāo) [CPU] 的值為 [90.0] 超過閾值,進(jìn)行告警!
[2023-07-11 22:05:34] 指標(biāo) [Memory] 的值為 [80.0] 超過閾值,進(jìn)行告警!

3、Java實(shí)現(xiàn)觀察者模式

package com.observer;

import java.util.List;

// ISubject通知者
public abstract class ISubject {

    List<IObserver> observerList;

    abstract void Register(IObserver observer);

    abstract void Remove(IObserver observer);

    abstract void Notify(Metric metric);
}

package com.observer;

import java.util.ArrayList;

// MetricSubject指標(biāo)通知者
public class MetricSubject extends ISubject {

    public MetricSubject() {
        this.observerList = new ArrayList<>();
    }

    @Override
    void Register(IObserver observer) {
        observerList.add(observer);
    }

    @Override
    void Remove(IObserver observer) {
        observerList.remove(observer);
    }

    @Override
    void Notify(Metric metric) {
        for (IObserver observer : observerList) {
            observer.Update(metric);
        }
    }
}

package com.observer;

// IObserver觀察者
public interface IObserver {

    void Update(Metric metric);
}

package com.observer;

// db觀察者
public class DbObserver implements IObserver {

    @Override
    public void Update(Metric metric) {
        String url = "[%s] 指標(biāo) [%s] 的值為 [%.1f] 超過閾值,保存到數(shù)據(jù)庫!";
        System.out.println(String.format(url, metric.Time, metric.Name, metric.Value));
    }
}

package com.observer;

// alert觀察者
public class AlertObserver implements IObserver {

    @Override
    public void Update(Metric metric) {
        String url = "[%s] 指標(biāo) [%s] 的值為 [%.1f] 超過閾值,進(jìn)行告警!";
        System.out.println(String.format(url, metric.Time, metric.Name, metric.Value));
    }
}

package com.observer;

import java.text.SimpleDateFormat;
import java.util.Date;

public class Test {
    public static void main(String[] args) {
        ISubject subject = new MetricSubject();
        IObserver observer1 = new DbObserver();
        IObserver observer2 = new AlertObserver();
        subject.Register(observer1);
        subject.Register(observer2);
        subject.Notify(new Metric("CPU", 90, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())));
        subject.Notify(new Metric("Memory", 80, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())));
        System.out.println("==========");
        // 后面不需要進(jìn)行持久化了
        subject.Remove(observer1);
        subject.Notify(new Metric("CPU", 90, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())));
        subject.Notify(new Metric("Memory", 80, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())));
    }
}

# 程序輸出
[2023-07-12 20:42:04] 指標(biāo) [CPU] 的值為 [90.0] 超過閾值,保存到數(shù)據(jù)庫!
[2023-07-12 20:42:04] 指標(biāo) [CPU] 的值為 [90.0] 超過閾值,進(jìn)行告警!
[2023-07-12 20:42:04] 指標(biāo) [Memory] 的值為 [80.0] 超過閾值,保存到數(shù)據(jù)庫!
[2023-07-12 20:42:04] 指標(biāo) [Memory] 的值為 [80.0] 超過閾值,進(jìn)行告警!
==========
[2023-07-12 20:42:04] 指標(biāo) [CPU] 的值為 [90.0] 超過閾值,進(jìn)行告警!
[2023-07-12 20:42:04] 指標(biāo) [Memory] 的值為 [80.0] 超過閾值,進(jìn)行告警!

4、使用Go實(shí)現(xiàn)EventBus

我們實(shí)現(xiàn)一個(gè)支持以下功能的事件總線：

1、異步不阻塞

2、支持任意參數(shù)值文章來源地址http://www.zghlxwxcb.cn/news/detail-594585.html

package eventbus

import (
	"fmt"
	"reflect"
	"sync"
)

// Bus
type Bus interface {
	Subscribe(topic string, handler interface{}) error
	Publish(topic string, args ...interface{})
}

// AsyncEventBus異步事件總線
type AsyncEventBus struct {
	handlers map[string][]reflect.Value
	lock     sync.Mutex
}

// NewAsyncEventBus
func NewAsyncEventBus() *AsyncEventBus {
	return &AsyncEventBus{
		handlers: map[string][]reflect.Value{},
		lock:     sync.Mutex{},
	}
}

// Subscribe訂閱
func (bus *AsyncEventBus) Subscribe(topic string, f interface{}) error {
	bus.lock.Lock()
	defer bus.lock.Unlock()
	v := reflect.ValueOf(f)
	if v.Type().Kind() != reflect.Func {
		return fmt.Errorf("handler is not a function")
	}
	handler, ok := bus.handlers[topic]
	if !ok {
		handler = []reflect.Value{}
	}
	handler = append(handler, v)
	bus.handlers[topic] = handler
	return nil
}

// Publish發(fā)布
// 這里異步執(zhí)行,并且不會等待返回結(jié)果
func (bus *AsyncEventBus) Publish(topic string, args ...interface{}) {
	handlers, ok := bus.handlers[topic]
	if !ok {
		fmt.Println("not found handlers in topic:", topic)
		return
	}
	params := make([]reflect.Value, len(args))
	for i, arg := range args {
		params[i] = reflect.ValueOf(arg)
	}
	for i := range handlers {
		go handlers[i].Call(params)
	}
}

package main

import (
	"fmt"
	. "proj/eventbus"
	"time"
)

func sub1(msg1, msg2 string) {
	time.Sleep(1 * time.Microsecond)
	fmt.Printf("sub1, %s %s\n", msg1, msg2)
}

func sub2(msg1, msg2 string) {
	fmt.Printf("sub2, %s %s\n", msg1, msg2)
}

func main() {
	bus := NewAsyncEventBus()
	bus.Subscribe("topic1", sub1)
	bus.Subscribe("topic1", sub2)
	bus.Publish("topic1", "1", "2")
	bus.Publish("topic1", "a", "b")
	time.Sleep(1 * time.Second)
}

# 程序輸出
sub2, a b
sub2, 1 2
sub1, 1 2
sub1, a b

到了這里，關(guān)于Go和Java實(shí)現(xiàn)觀察者模式的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！