观察者模式
目录
- 观察者模式
-
-
- 分析
-
- 使用场景
- 代码实现
- 实例
-
- 基础实现
-
- 代码
- 单元测试
- 使用 Golang 实现 EventBus
-
- 代码
- 单元测试
- 结果
- 总结
观察者模式的应用场景非常广泛，小到代码层面的解耦，大到架构层面的系统解耦，再或者一些产品的设计思路，都有这种模式的影子。
观察者模式：观察者模式定义了一种一对多的依赖关系，让多个观察者对象同时监听某一个主题对象。这个主题对象在状态发生变化时，会通知所有观察者对象，使他们能够自动更新自己。
分析
观察者模式算是比较通用的设计模式了，思路也比较清晰。Subject里有Observer集合，当Subject有更新时，通知各个Observer。Subject为了管理Observer，自身设置了增加、删除功能。
当Subject有更新，通知Observer时，有很多细节值得讨论。
第一个问题是使用同步阻塞还是异步非阻塞
- 同步阻塞是最经典的实现方式，主要是为了代码解耦；
- 异步非阻塞除了能实现代码解耦之外，还能提高代码的执行效率；
第二个问题是如何保证所有Observer都通知成功。
- 方案一是利用消息队列ACK的能力，Observer订阅消息队列。Subject只需要确保信息通知给消息队列即可。
- 方案二是Subject将失败的通知记录，方便后面进行重试。
- 方案三是定义好规范，例如只对网络失败这种错误进行记录，业务失败类型不管理，由业务自行保证成功。
第三个问题是不同进程/系统如何进行通知。
- 进程间的观察者模式解耦更加彻底，一般是基于消息队列来实现，用来实现不同进程间的被观察者和观察者之间的交互。
使用场景
观察者模式的使用场景还是很多的，它和享元模式一样，更多体现的是一种设计理念。
商家购买服务的订单，有多个状态，如交易成功、交易取消、开始履约等。业务场景只关注交易成功、交易取消和开始履约。如果不使用观察者模式，可以用if-else解决这个问题。但各个状态的处理逻辑比较复杂，而且状态有可能存在乱序的情况，导致要处理的case更多了。这时候用观察者模式就很合适，订单状态变更为Subject，具体处理逻辑为Observer，状态变更时，通知所有处理逻辑，谁合适处理便由谁处理。不但隔离性变好，扩展性也增强了。
代码实现
package main
import "fmt"
type PurchaseOperFunc func(status string, data string) (res bool, err error)
/**
* @Description: 注册的观察者
*/
var PurchaseOperFuncArr = []PurchaseOperFunc{
create,
isDeleted,
apply,
}
/**
* @Description: 用于创建的观察者
* @param status
* @param data
* @return res
* @return err
*/
func create(status string, data string) (res bool, err error) {
if status == "create" {
fmt.Println("开始创建")
return true, nil
}
return true, nil
}
/**
* @Description: 用于删除的观察者
* @param status
* @param data
* @return res
* @return err
*/
func isDeleted(status string, data string) (res bool, err error) {
if status == "delete" {
fmt.Println("开始删除")
return true, nil
}
return true, nil
}
/**
* @Description: 用于履约的观察者
* @param status
* @param data
* @return res
* @return err
*/
func apply(status string, data string) (res bool, err error) {
if status == "apply" {
fmt.Println("开始履约")
return true, nil
}
return true, nil
}
func main() {
status := "create"
data := "订单数据"
//有状态更新时，通知所有观察者
for _, oper := range PurchaseOperFuncArr {
res, err := oper(status, data)
if err != nil {
fmt.Println("操作失败")
break
}
if res == false {
fmt.Println("处理失败")
break
}
}
}
输出：
*myproject go run main.go
开始创建
这个代码是个简单版，但整体逻辑是一样的。这种写法的好处有如下几个：
1、 如果要处理新的状态，则实现后再PurchaseOperFuncArr添加即可，改动不大；
2、 根据具体业务来说，只能有一个逻辑是真正负责执行的，所以有一个错误就报错，然后重试即可不过具体逻辑需要做好幂等；
实例
基础实现
代码
package observer
import "fmt"
// ISubject subject
type ISubject interface {
Register(observer IObserver)
Remove(observer IObserver)
Notify(msg string)
}
// IObserver 观察者
type IObserver interface {
Update(msg string)
}
// Subject Subject
type Subject struct {
observers []IObserver
}
// Register 注册
func (sub *Subject) Register(observer IObserver) {
sub.observers = append(sub.observers, observer)
}
// Remove 移除观察者
func (sub *Subject) Remove(observer IObserver) {
for i, ob := range sub.observers {
if ob == observer {
sub.observers = append(sub.observers[:i], sub.observers[i+1:]...)
}
}
}
// Notify 通知
func (sub *Subject) Notify(msg string) {
for _, o := range sub.observers {
o.Update(msg)
}
}
// Observer1 Observer1
type Observer1 struct{
}
// Update 实现观察者接口
func (Observer1) Update(msg string) {
fmt.Printf("Observer1: %s", msg)
}
// Observer2 Observer2
type Observer2 struct{
}
// Update 实现观察者接口
func (Observer2) Update(msg string) {
fmt.Printf("Observer2: %s", msg)
}
单元测试
package observer
import "testing"
func TestSubject_Notify(t *testing.T) {
sub := &Subject{
}
sub.Register(&Observer1{
})
sub.Register(&Observer2{
})
sub.Notify("hi")
}
使用 Golang 实现 EventBus
我们实现一个支持以下功能的事件总线
1、 异步不阻塞；
2、 支持任意参数值；
代码
package eventbus
import (
"fmt"
"reflect"
"sync"
)
// Bus 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 new
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 发布
// 这里异步执行，并且不会等待返回结果
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 eventbus
import (
"fmt"
"testing"
"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 TestAsyncEventBus_Publish(t *testing.T) {
bus := NewAsyncEventBus()
bus.Subscribe("topic:1", sub1)
bus.Subscribe("topic:1", sub2)
bus.Publish("topic:1", "test1", "test2")
bus.Publish("topic:1", "testA", "testB")
time.Sleep(1 * time.Second)
}
结果
=== RUN TestAsyncEventBus_Publish
sub2, testA testB
sub2, test1 test2
sub1, testA testB
sub1, test1 test2
--- PASS: TestAsyncEventBus_Publish (1.01s)
总结
实际上，设计模式要干的事情就是解耦。创建型模式是将创建和使用代码解耦，结构型模式是将不同功能代码解耦，行为型模式是将不同的行为代码解耦，具体到观察者模式，它是将观察者和被观察者代码解耦。
版权声明：本文不是「本站」原创文章，版权归原作者所有 | 原文地址: