Go设计模式13-观察者模式(实现简单的EventBus)

笔记

代码实现

基础实现

Code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
package observer

import "fmt"

// ISubject subject
type ISubject interface {
Register(observer IObsever)
Remove(observer IObsever)
Notify(observer IObsever)
}

// IObsever 观察者
type IObsever interface {
Update(msg string)
}

// Subject Subject
type Subject struct {
observers []IObsever
}

// Register 注册
func (sub *Subject) Register(observer IObsever) {
sub.observers = append(sub.observers, observer)
}

// Remove 移除观察者
func (sub *Subject) Remove(observer IObsever) {
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)
}
}

// Obsever1 Obsever1
type Obsever1 struct{}

// Update 实现观察者接口
func (Obsever1) Update(msg string) {
fmt.Printf("Obsever1: %s", msg)
}

// Obsever2 Obsever2
type Obsever2 struct{}

// Update 实现观察者接口
func (Obsever2) Update(msg string) {
fmt.Printf("Obsever2: %s", msg)
}

单元测试

1
2
3
4
5
6
7
8
9
10
package observer

import "testing"

func TestSubject_Notify(t *testing.T) {
sub := &Subject{}
sub.Register(&Obsever1{})
sub.Register(&Obsever2{})
sub.Notify("hi")
}

使用 Golang 实现 EventBus

我们实现一个支持以下功能的事件总线

  1. 异步不阻塞
  2. 支持任意参数值

Code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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)
}
}

单元测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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)
}

结果

1
2
3
4
5
6
=== RUN   TestAsyncEventBus_Publish
sub2, testA testB
sub2, test1 test2
sub1, testA testB
sub1, test1 test2
--- PASS: TestAsyncEventBus_Publish (1.01s)

本博客所有文章除特别声明外,均采用 CC BY-SA 4.0 协议,转载请注明出处,禁止全文转载