如何使用Go语言构建高效低延迟的事件循环

package eventloop

import (
    "fmt"
    "sync"
    "time"
)

// EventLoop 结构体定义了事件循环的核心组件
type EventLoop struct {
    nextFunc chan func() // 用于顺序执行的任务通道
    curFunc  chan func() // 用于当前tick内并发执行的任务通道
    quit     chan struct{} // 用于通知事件循环退出的通道
    wg       sync.WaitGroup // 用于等待所有并发任务完成
}

// NewEventLoop 创建并初始化一个新的EventLoop实例
func NewEventLoop() *EventLoop {
    el := &EventLoop{
        // 通道容量可根据实际需求调整，避免过度缓冲或阻塞
        nextFunc: make(chan func(), 10), // 示例容量
        curFunc:  make(chan func(), 10),  // 示例容量
        quit:     make(chan struct{}),
    }
    go el.eventLoopRunner() // 启动事件循环的后台goroutine
    return el
}

// NextTick 提交一个任务，该任务将在下一个“tick”中顺序执行
func (el *EventLoop) NextTick(f func()) {
    select {
    case el.nextFunc <- f:
        // 任务成功发送
    case <-el.quit:
        // 事件循环已退出，不再接受新任务
        fmt.Println("EventLoop is quitting, NextTick task rejected.")
    }
}

// CurrentTick 提交一个任务，该任务将在当前“tick”内并发执行
func (el *EventLoop) CurrentTick(f func()) {
    select {
    case el.curFunc <- f:
        // 任务成功发送
    case <-el.quit:
        // 事件循环已退出，不再接受新任务
        fmt.Println("EventLoop is quitting, CurrentTick task rejected.")
    }
}

// Quit 通知事件循环退出，并等待所有未完成的任务执行完毕
func (el *EventLoop) Quit() {
    close(el.quit) // 关闭quit通道，通知eventLoopRunner退出
    // 关闭输入通道，防止新的任务进入，并让eventLoopRunner的select语句退出
    close(el.nextFunc)
    close(el.curFunc)
    // 等待所有在eventLoopRunner中启动的goroutine完成
    // 注意：这里的wg只用于eventLoopRunner本身，如果CurrentTick内部的f()也启动goroutine，需要f()内部自行管理
}

// eventLoopRunner 是事件循环的核心逻辑，在一个独立的goroutine中运行
func (el *EventLoop) eventLoopRunner() {
    for {
        select {
        case f, ok := <-el.nextFunc:
            if !ok {
                // nextFunc 已关闭，表示需要退出循环
                fmt.Println("NextTick channel closed. Exiting eventLoopRunner.")
                return
            }
            // 执行顺序任务
            f()

            // 处理当前tick中的所有并发任务
            el.drainCurrentTickTasks()

        case <-el.quit:
            // 收到退出信号，尝试清空剩余任务并退出
            fmt.Println("Quit signal received. Draining remaining tasks...")
            // 尝试处理可能仍在通道中的任务
            el.drainRemainingTasks()
            return
        }
    }
}

// drainCurrentTickTasks 负责处理当前tick中所有通过curFunc提交的并发任务
func (el *EventLoop) drainCurrentTickTasks() {
    // 使用一个内部循环和select-default模式，非阻塞地从curFunc通道中读取任务
    // 这种模式确保了在没有更多curFunc任务时，循环立即退出，避免阻塞
    drain:
    for {
        select {
        case f := <-el.curFunc:
            el.wg.Add(1) // 增加WaitGroup计数
            go func(task func()) {
                defer el.wg.Done() // 任务完成后减少WaitGroup计数
                task()
            }(f)
        default:
            // curFunc 通道当前没有可读数据，退出drain循环
            break drain
        }
    }
    // 等待所有当前tick中启动的并发任务完成
    el.wg.Wait()
    fmt.Println("All CurrentTick tasks for this tick completed.")
}

// drainRemainingTasks 在退出时尝试处理通道中剩余的任务
func (el *EventLoop) drainRemainingTasks() {
    // 优先处理nextFunc中剩余的顺序任务
    for {
        select {
        case f := <-el.nextFunc:
            f()
        default:
            goto drainCurFunc // nextFunc已空，转而处理curFunc
        }
    }

drainCurFunc:
    // 处理curFunc中剩余的并发任务
    for {
        select {
        case f := <-el.curFunc:
            el.wg.Add(1)
            go func(task func()) {
                defer el.wg.Done()
                task()
            }(f)
        default:
            goto endDrain // curFunc已空
        }
    }

endDrain:
    el.wg.Wait() // 等待所有在退出过程中启动的goroutine完成
    fmt.Println("All remaining tasks drained.")
}

// 示例用法
func main() {
    el := NewEventLoop()

    // 提交NextTick任务
    el.NextTick(func() {
        fmt.Println("NextTick task 1: Sequential operation.")
    })

    // 提交CurrentTick任务，它们将在NextTick任务1之后并发执行
    el.CurrentTick(func() {
        time.Sleep(100 * time.Millisecond) // 模拟耗时操作
        fmt.Println("CurrentTick task A: Concurrent operation.")
    })
    el.CurrentTick(func() {
        time.Sleep(50 * time.Millisecond)
        fmt.Println("CurrentTick task B: Concurrent operation.")
    })

    el.NextTick(func() {
        fmt.Println("NextTick task 2: Sequential operation after all current tick tasks.")
    })

    // 提交更多CurrentTick任务
    el.CurrentTick(func() {
        fmt.Println("CurrentTick task C: Another concurrent operation.")
    })

    // 模拟程序运行一段时间
    time.Sleep(500 * time.Millisecond)

    // 优雅退出事件循环
    fmt.Println("\nQuitting event loop...")
    el.Quit()
    // 确保所有任务完成，主goroutine可以等待eventLoopRunner退出
    // 在实际应用中，可能需要更复杂的同步机制来确保所有任务完成
    time.Sleep(1 * time.Second) // 给予足够时间让退出流程完成
    fmt.Println("Program finished.")
}