Docker组件go-event 源码学习

go-event是一个在Docker项目中使用到的一个事件分发组件,实现了常规的广播,队列等事件分发模型,代码简洁明了,也适合初学者对Go语言的入门,对channel用来同步,通信也会加深理解。



核心数据结构
Event



type Event interface{}
Event被封装为一个空接口,接受任意类型。在go-events表示一个可以被分发的事件。



interface{}的底层类似于c语言中的void,但比void强大很多,比如interface{}保存了指向对象的指针和类型,而c程序员使用void*时,必须自己去保证对象的类型是正确的)



Sink



type Sink interface {
Write(event Event) error
Close() error
}
Sink是一个用来分发事件(Event)的结构。可以当作事件的处理者,使用接口的方式声明。只要对象实现了这两个方法,就可以被当作一个Sink。
核型方法



Write(event Event) error



定义了事件如何被分发的策略。
Close() error



当Sink被关闭的处理策略。
https://studygolang.com/articles/30112



go-event核心就是围绕Sink做文章,docker官方给出了一个http的例子,就是当调用Write时,发起一次post请求。:



func (h *httpSink) Write(event Event) error {
p, err := json.Marshal(event)
if err != nil {
return err
}
body := bytes.NewReader(p)
resp, err := h.client.Post(h.url, “application/json”, body)
if err != nil {
return err
}
defer resp.Body.Close()



if resp.Status != 200 {
return errors.New("unexpected status")
}

return nil }


// implement (*httpSink).Close()
实现模型
到此为止,sink定义了事件分发的基本单位。在go-event中,封装了广播,消息队列两种消息分发的模型,具体来说,就是实现了Sink接口的两个结构体。



Boadcaster
type Broadcaster struct {
sinks []Sink //所包含的Sink
events chan Event// 同步Event的channel
adds chan configureRequest //adds和remove必须保证thread-safe,所以采用channel同步
removes chan configureRequest



shutdown chan struct{}
closed chan struct{}
once sync.Once } Boardcaster由多个Sink组成,当Boardcaster接收到一个事件时,会调用自身包含的所有Sink的Write()方法 go-events设计之初就实现协程之间的消息分发,需要保证thread-safe,所以对event的处理,添加,移除Sink都使用管道来通信。这也是Go的一个使用原则:


使用通信来共享内存,而不是通过共享内存来通信



在Broadcaster中所有的临界资源(sinks,event)都通过自身的run()函数统一管理,外界则通过相应的channel 同步给Broadcaster
例如Write()



func (b *Broadcaster) Write(event Event) error {
select {
case b.events <- event:
case <-b.closed:
return ErrSinkClosed
}
return nil
}
可以看到增减sink都是通过向对应的channel写入数据进行的。



func (b *Broadcaster) Add(sink Sink) error {
return b.configure(b.adds, sink) // will be block until ch can be writen
}



func (b *Broadcaster) configure(ch chan configureRequest, sink Sink) error {
response := make(chan error, 1)



for {
select {
case ch <- configureRequest{
sink: sink,
response: response}:
ch = nil // ?
case err := <-response:
return err
case <-b.closed:
return ErrSinkClosed
}
} } 核心run函数的实现,监听Boardcast管道上的相应事件,并作出处理。


func (b *Broadcaster) run() {
defer close(b.closed)
//将remove封装了一下,因为下面两处都会用到
remove := func(target Sink) {
for i, sink := range b.sinks {
if sink == target {
b.sinks = append(b.sinks[:i], b.sinks[i+1:]…)
break
}
}
}
// 轮训处理channel上的事件
for {
select {
case event := <-b.events: //有事件到来,进行广播
for _, sink := range b.sinks {
if err := sink.Write(event); err != nil {
if err == ErrSinkClosed {
// remove closed sinks
remove(sink)
continue
}
logrus.WithField(“event”, event).WithField(“events.sink”, sink).WithError(err).
Errorf(“broadcaster: dropping event”)
}
}
case request := <-b.adds: //增加sink事件
// while we have to iterate for add/remove, common iteration for
// send is faster against slice.



        var found bool
for _, sink := range b.sinks {
if request.sink == sink {
found = true
break
}
}

if !found {
b.sinks = append(b.sinks, request.sink)
}
// b.sinks[request.sink] = struct{}{}
request.response <- nil // 唤醒阻塞的configure()函数

case request := <-b.removes://删除sink事件
remove(request.sink)
request.response <- nil
case <-b.shutdown:
// close all the underlying sinks
for _, sink := range b.sinks {
if err := sink.Close(); err != nil && err != ErrSinkClosed {
logrus.WithField("events.sink", sink).WithError(err).
Errorf("broadcaster: closing sink failed")
}
}
return
}
} } queue queue使用contaienr/list实现了典型的生产消费者模型


type Queue struct {
dst Sink
events *list.List
cond *sync.Cond
mu sync.Mutex
closed bool
}
核心函数run(),在队列中取出下一个event,交给自身的sink处理,在没有事件队列的情况下,eq.next()总是阻塞的(使用条件变量进行同步)



func (eq *Queue) run() {
for {
event := eq.next()



    if event == nil {
return // nil block means event queue is closed.
}

if err := eq.dst.Write(event); err != nil {
logrus.WithFields(logrus.Fields{
"event": event,
"sink": eq.dst,
}).WithError(err).Debug("eventqueue: dropped event")
}
} } 生产者:q.next() 消费者:write()


func (eq *Queue) Write(event Event) error {
eq.mu.Lock()
defer eq.mu.Unlock()



if eq.closed {
return ErrSinkClosed
}

eq.events.PushBack(event)
eq.cond.Signal() // signal waiters

return nil }


func (eq *Queue) next() Event {
eq.mu.Lock()
defer eq.mu.Unlock()



for eq.events.Len() < 1 {
if eq.closed {
eq.cond.Broadcast()
return nil
}

eq.cond.Wait()
}

front := eq.events.Front()
block := front.Value.(Event)
eq.events.Remove(front)

return block }


Category golang