使用Go语言构建低延迟事件循环与任务同步机制

package eventloop

import "fmt"

// EventLoop 结构体定义了事件循环的核心组件
type EventLoop struct {
    nextFunc chan func() // 用于接收顺序执行的“下一刻”任务
    curFunc  chan func() // 用于接收与当前周期相关的“当前刻”任务
}

// NewEventLoop 创建并初始化一个新的EventLoop实例
func NewEventLoop() *EventLoop {
    // 根据实际需求调整通道容量，合理容量可以减少阻塞，提高吞吐量
    el := &EventLoop{
        nextFunc: make(chan func(), 10), // 示例容量为10
        curFunc:  make(chan func(), 10),  // 示例容量为10
    }
    // 启动一个独立的goroutine来运行事件循环的核心逻辑
    go eventLoop(el)
    return el
}

// NextTick 提交一个“下一刻”任务。这些任务将按提交顺序执行。
func (el *EventLoop) NextTick(f func()) {
    el.nextFunc <- f
}

// CurrentTick 提交一个“当前刻”任务。这些任务将在当前NextTick任务完成后，
// 且在下一个NextTick任务开始前，被全部处理。
func (el *EventLoop) CurrentTick(f func()) {
    el.curFunc <- f
}

// Quit 安全地关闭事件循环。通过关闭nextFunc通道，通知eventLoop goroutine退出。
func (el *EventLoop) Quit() {
    close(el.nextFunc)
    // 注意：如果curFunc中仍有未处理的任务，它们将不会被执行。
    // 根据具体需求，可能需要在此处添加额外的清理或等待机制。
}

// eventLoop 是事件循环的核心处理逻辑，在一个独立的goroutine中运行。
func eventLoop(el *EventLoop) {
    for {
        // 1. 等待并执行一个NextTick任务
        // 从nextFunc通道接收任务，如果通道关闭且无数据，ok为false，循环退出。
        f, ok := <-el.nextFunc
        if !ok {
            fmt.Println("EventLoop: nextFunc channel closed, exiting.")
            return // nextFunc通道关闭，事件循环优雅退出
        }
        f() // 执行当前的NextTick任务

        // 2. 耗尽所有当前可用的CurrentTick任务
        // 这是一个关键的“等待”机制：在进入下一个NextTick任务之前，
        // 确保所有在当前周期内提交的CurrentTick任务都被处理。
    drain:
        for {
            select {
            case f := <-el.curFunc:
                // 如果curFunc有任务，立即执行
                f()
            default:
                // 如果curFunc当前没有任务，立即跳出drain循环
                // 避免阻塞，实现低延迟检查
                break drain
            }
        }
    }
}

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

    // 提交一个NextTick任务
    el.NextTick(func() {
        fmt.Println("--- NextTick 1 Started ---")
        // 在NextTick 1执行期间提交CurrentTick任务
        el.CurrentTick(func() { fmt.Println("CurrentTick A") })
        el.CurrentTick(func() { fmt.Println("CurrentTick B") })
        fmt.Println("--- NextTick 1 Finished ---")
    })

    // 提交另一个NextTick任务
    el.NextTick(func() {
        fmt.Println("--- NextTick 2 Started ---")
        el.CurrentTick(func() { fmt.Println("CurrentTick C") })
        fmt.Println("--- NextTick 2 Finished ---")
    })

    // 提交一个NextTick任务，但其CurrentTick任务可能在它执行后才被提交
    el.NextTick(func() {
        fmt.Println("--- NextTick 3 Started ---")
        fmt.Println("--- NextTick 3 Finished ---")
    })
    el.CurrentTick(func() { fmt.Println("CurrentTick D (submitted after NextTick 3 started)") }) // D会在NextTick 3之后，NextTick 4之前被处理

    // 模拟程序运行一段时间
    // 在实际应用中，这里可能是主goroutine的其他工作，或者等待信号
    // 为了演示效果，我们简单等待一段时间
    // time.Sleep(time.Second) // 实际使用时，可能需要更复杂的等待机制

    // 提交最后一个NextTick任务，并模拟退出
    el.NextTick(func() {
        fmt.Println("--- NextTick 4 Started ---")
        el.CurrentTick(func() { fmt.Println("CurrentTick E") })
        fmt.Println("--- NextTick 4 Finished ---")
    })

    // 安全关闭事件循环
    el.Quit()

    // 等待事件循环goroutine完全退出
    // 在实际应用中，可能需要一个sync.WaitGroup来确保所有任务完成
    select {} // 阻塞主goroutine，等待eventLoop goroutine输出退出信息
}