Go语言实现-观察者模式

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

前前言

这个类经过我的正式投入使用啊，发现不对劲，这样做可能会导致线程死锁

比如你dispatch一个event，然后在这个回调里把那个事件的侦听给remove掉了，那么就会导致线程死锁（这个问题找了好久啊，刚刚调试的时候才发现了）

还有就是获取func的引用的问题，golang那半c半java的语法，我改用了新的方法

源码已经修改！

前言：

呀，学Go语言两周了，感觉上手挺快的，golang虽然和c语言很想，但是避免掉了很多指针相关的东东，所以学起来特别轻松。

但是途中坎坷颇多啊，资料是少之又少啊，搜索引擎都搜不到啥东西，唯有golang.org和github的ebook上才有比较完整的资料，加了个golang群么，大牛都潜水不说话，然后我瞎了，只能自己琢磨。

也是自己闹着玩吧，最近想写一个服务器（目前已经能同步移动了），我也想学一门后端语言，于是就选了google的go语言了，听说并发性能挺好的。

然而，golang这语言貌似都是被用作web服务开发了的，群里一般都是在讨论web开发的问题，而我是做socket开发的，当然，golang里面不叫socket。

工作比较忙，只有下班后那点时间来学golang了，学得比较基础，大牛就当走走场好了。

正题：

我是一名页游前端开发人员，当然我是as3开发者，对as3的观察者模式-事件机制，那是太依赖了，而golang里面原生并不提供这种机制。

golang有的只是十分相似的goroutine，也就是底层支持的并发机制，然后线程间通讯就是channel，好比于as3中的Event，当然不能直接比较，要封装过。

观察者模式就是指一对多的依赖关系，生产者分派消息，消费者全都能收到消息（全局观察模式），这样，可以降低模块间的耦合度，我们要做的，就是来管理这三者。

然后，用golang来实现这一设计模式是很简单的，我仅用了一百多行，就简单地实现了，直接看代码吧：

  1 package tbs
  2 
  3 import (
  4     //"fmt"
  5     "unsafe"
  6 )
  7 
  8 type Dispatcher struct {
  9     listeners map[string]*EventChain
 10 }
 11 
 12 type EventChain struct {
 13     chs       []chan *Event
 14     callbacks []*EventCallback
 15 }
 16 
 17 func createEventChain() *EventChain {
 18     return &EventChain{chs: []chan *Event{}, callbacks: []*EventCallback{}}
 19 }
 20 
 21 type Event struct {
 22     eventName string
 23     Params    map[string]interface{}
 24 }
 25 
 26 func CreateEvent(eventName string, params map[string]interface{}) *Event {
 27     return &Event{eventName: eventName, Params: params}
 28 }
 29 
 30 type EventCallback func(*Event)
 31 
 32 var _instance *Dispatcher
 33 
 34 func SharedDispatcher() *Dispatcher {
 35     if _instance == nil {
 36         _instance = &Dispatcher{}
 37         _instance.Init()
 38     }
 39 
 40     return _instance
 41 }
 42 
 43 func (this *Dispatcher) Init() {
 44     this.listeners = make(map[string]*EventChain)
 45 }
 46 
 47 func (this *Dispatcher) AddEventListener(eventName string, callback *EventCallback) {
 48     eventChain, ok := this.listeners[eventName]
 49     if !ok {
 50         eventChain = createEventChain()
 51         this.listeners[eventName] = eventChain
 52     }
 53 
 54     exist := false
 55     //fmt.Println("add len:", len(eventChain.callbacks))
 56     for _, item := range eventChain.callbacks {
 57         a := *(*int)(unsafe.Pointer(item))
 58         b := *(*int)(unsafe.Pointer(callback))
 59         //fmt.Println("add", a, b)
 60         if a == b {
 61             exist = true
 62             break
 63         }
 64     }
 65 
 66     if exist {
 67         return
 68     }
 69 
 70     ch := make(chan *Event)
 71 
 72     eventChain.chs = append(eventChain.chs[:], ch)
 73     eventChain.callbacks = append(eventChain.callbacks[:], callback)
 74 
 75     go this.handler(eventName, ch, callback)
 76 }
 77 
 78 func (this *Dispatcher) handler(eventName string, ch chan *Event, callback *EventCallback) {
 79     //fmt.Printf("add listener: %s\n", eventName)
 80     //fmt.Println("chan: ", ch)
 81     for {
 82         event := <-ch
 83         //fmt.Println("event out:", eventName, event, ch)
 84         if event == nil {
 85             break
 86         }
 87         go (*callback)(event)
 88     }
 89 }
 90 
 91 func (this *Dispatcher) RemoveEventListener(eventName string, callback *EventCallback) {
 92     eventChain, ok := this.listeners[eventName]
 93     if !ok {
 94         return
 95     }
 96 
 97     var ch chan *Event
 98     exist := false
 99     key := 0
100     for k, item := range eventChain.callbacks {
101         a := *(*int)(unsafe.Pointer(item))
102         b := *(*int)(unsafe.Pointer(callback))
103         //fmt.Println("remove", a, b)
104         if a == b {
105             exist = true
106             ch = eventChain.chs[k]
107             key = k
108             break
109         }
110     }
111 
112     if exist {
113         //fmt.Printf("remove listener: %s\n", eventName)
114         //fmt.Println("chan: ", ch)
115         ch <- nil
116 
117         eventChain.chs = append(eventChain.chs[:key], eventChain.chs[key+1:]...)
118         eventChain.callbacks = append(eventChain.callbacks[:key], eventChain.callbacks[key+1:]...)
119         //fmt.Println(len(eventChain.chs))
120     }
121 }
122 
123 func (this *Dispatcher) DispatchEvent(event *Event) {
124     eventChain, ok := this.listeners[event.eventName]
125     if ok {
126         ////fmt.Printf("dispatch event: %s\n", event.eventName)
127         for _, chEvent := range eventChain.chs {
128             chEvent <- event
129         }
130     }
131 }

这个类里定义了三个结构，Dispatcher:分派器主类，Event:事件类，EventChain:事件链类

如果你要使用这个类，那你只要那Dispatcher的单例方法：

SharedDispatcher()

来进行操作好了

要创建Event，你是要使用创建方法

CreateEvent(eventNamestring,paramsmap[string]interface{})

来创建

当然，demo还得贴上

 1 package main
 2 
 3 import (
 4     "fmt"
 5     "tbs"
 6     "time"
 7 )
 8 
 9 type MClass struct {
10     dispatcher tbs.Dispatcher
11 }
12 
13 func main() {
14     mc := &MClass{}
15     mc.Start()
16 }
17 
18 func (this *MClass) Start() {
19     //获取分派器单例
20     dispatcher := tbs.SharedDispatcher()
21 
22     //添加监听1
23     var fun1 tbs.EventCallback = this.onTest
24     dispatcher.AddEventListener("test", &fun1)
25 
26     //再添加监听2
27     var fun2 tbs.EventCallback = this.onTest2
28     dispatcher.AddEventListener("test", &fun2)
29 
30     //随便弄个事件携带的参数，我把参数定义为一个map
31     params := make(map[string]interface{})
32     params["id"] = 1000
33     //创建一个事件对象
34     event := tbs.CreateEvent("test", params)
35     //把事件分派出去
36     dispatcher.DispatchEvent(event)
37 
38     //移除监听1
39     dispatcher.RemoveEventListener("test", &fun1)
40 
41     //再把事件分派出去一次
42     dispatcher.DispatchEvent(event)
43 
44     //因为主线程不会等子线程而直接关闭进程，这样会看不到效果，所以我在这里加了阻塞式延时
45     time.Sleep(time.Second * 1)
46 }
47 
48 //回调出得到的就是一个event对象了
49 func (this *MClass) onTest(event *tbs.Event) {
50     fmt.Println("onTest", event.Params["id"])
51 }
52 
53 func (this *MClass) onTest2(event *tbs.Event) {
54     fmt.Println("onTest2", event.Params["id"])
55 }

输出结果：

add listener: test
add listener: test
dispatch event: test
onTest 1000
remove listener: test
dispatch event: test
onTest2 1000
onTest2 1000
成功: 进程退出代码 0.

哈哈，成功地运行了。

demo你面的注释已经非常详尽了，看不懂就在下面问我好了！

昨晚我拿他来封装了一下golang的socket，改成了事件驱动，耦合度瞬间降低了很多。

 1 func onServerStarted(event *tbs.Event) {
 2     fmt.Println("server started.")
 3 }
 4 
 5 func onAccept(event *tbs.Event) {
 6     socket := (event.Params["socket"]).(*tbs.Socket)
 7 
 8     fmt.Printf("client[#%d] connect on:%s\n", socket.Sign, socket.Conn.RemoteAddr().String())
 9 }
10 
11 func onData(event *tbs.Event) {
12     socket := (event.Params["socket"]).(*tbs.Socket)
13     bytes := (event.Params["bytes"]).([]byte)
14 
15     fmt.Printf("[#%d]:", socket.Sign)
16     fmt.Println(bytes)
17 }
18 
19 func onClosed(event *tbs.Event) {
20     socket := (event.Params["socket"]).(*tbs.Socket)
21     fmt.Printf("[#%d] closed\n", socket.Sign)
22 }