event_loop.go

// 用于演示go里面如何动态向select添加channel
// 同时演示了网络编程框架里面EventLoop大概长啥样

package main

import (
	"fmt"
	"reflect"
	"time"
)

type EventLoop struct {
	running bool
	// 用于唤醒loop线程，同时在loop线程执行一些内部任务
	wakeupch chan func()
	// 所有的已注册channel集合
	channels []reflect.SelectCase
}

func NewEventLoop() *EventLoop {
	e := &EventLoop{
		wakeupch: make(chan func(), 10),
	}
	e.addChannel(e.wakeupch)
	return e
}

func (e *EventLoop) addChannel(ch chan func()) {
	e.channels = append(e.channels, reflect.SelectCase{
		Dir:  reflect.SelectRecv,
		Chan: reflect.ValueOf(ch),
	})
}

func (e *EventLoop) removeChannel(idx int) {
	e.channels = append(e.channels[:idx], e.channels[idx+1:]...)
}

// RunInLoop 用于在event loop上下文里面执行一个函数
func (e *EventLoop) RunInLoop(f func()) {
	e.wakeupch <- f
}

func (e *EventLoop) Close() {
	e.RunInLoop(func() {
		e.running = false
	})
}

// 向EventLoop注册事件源
func (e *EventLoop) AddChannel(ch chan func()) {
	e.RunInLoop(func() {
		e.addChannel(ch)
	})
}

func (e *EventLoop) Loop() {
	e.running = true
	for e.running {
		idx, event, ok := reflect.Select(e.channels)
		// ok指示channel是否没有关闭，跟v, ok <- ch的ok一个意思
		if !ok {
			e.removeChannel(idx)
			continue
		}
		event.Call(nil)
	}
}

var global int

func runEventSource(no int, e *EventLoop) {
	ch := make(chan func())
	e.AddChannel(ch)
	for i := 0; i < 3; i++ {
		eventno := i
		ch <- func() {
			fmt.Printf("event from source %d, event number %d\n", no, eventno)
		}
		// 在event loop里面修改global，不用加锁
		e.RunInLoop(func() {
			global++
			fmt.Printf("global: %d\n", global)
		})
	}
	close(ch)
}

func runTimerSource(e *EventLoop) {
	ch := make(chan func())
	tick := time.NewTicker(time.Second)
	e.AddChannel(ch)
	for t := range tick.C {
		t := t
		ch <- func() {
			fmt.Printf("event from timer source %s\n", t)
		}
	}
}

func start(e *EventLoop) {
	go runTimerSource(e)
	for i := 0; i < 3; i++ {
		go runEventSource(i, e)
	}
}

func main() {
	eventLoop := NewEventLoop()
	go start(eventLoop)
	eventLoop.Loop()
}