事件循环中的并发任务管理挑战
在设计一个事件循环（Event Loop）时，我们常常需要处理两种类型的任务：一种是需要按序执行的常规任务，另一种是在某个特定“tick”内可以并发执行，但必须在该“tick”结束前全部完成的任务。传统的解决方案可能包括：
- 使用互斥锁（Mutex）和计数器： 维护一个活跃线程/goroutine的计数器，通过互斥锁保护其增减。主循环通过不断检查计数器是否归零来判断所有任务是否完成。这种方法的问题在于，持续检查计数器会导致CPU空转（busy-waiting），严重浪费计算资源。
- 使用定时器（time.Sleep）： 主循环周期性地暂停，然后检查计数器。这种方法虽然避免了CPU空转，但引入了固定的延迟，无法实现低延迟的任务完成检测。
理想的解决方案应该既能确保所有并发任务完成，又能避免CPU空转和不必要的延迟。Go语言的通道机制为解决此类问题提供了强大的原语。
基于通道的事件循环设计
Go语言的通道（channels）是协程（goroutine）之间通信的推荐方式，它天然地支持并发安全和同步。我们可以利用通道的阻塞特性来优雅地实现事件循环中的任务调度与等待。
以下是一个改进的Go事件循环实现，它通过两个通道来区分不同类型的任务：
立即学习“go语言免费学习笔记（深入）”；
- nextFunc：用于接收需要顺序执行的任务。
- curFunc：用于接收在当前“tick”内可以并发执行的任务。
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.")
}设计解析与注意事项
-
通道作为任务队列：
- nextFunc 和 curFunc 作为带缓冲的通道，充当了任务队列。任务通过 NextTick 和 CurrentTick 方法发送到相应的通道。
- 缓冲通道允许在事件循环处理任务的同时，外部可以异步地提交任务，提高并发度。容量的选择应根据预期的任务提交速率和处理速率进行调整。
-
eventLoopRunner 核心逻辑：
- 在一个独立的goroutine中运行，通过 select 语句监听 nextFunc 和 quit 通道。
- 当从 nextFunc 接收到任务时，它首先执行该顺序任务。
- 紧接着，调用 drainCurrentTickTasks() 方法来处理所有在当前“tick”内提交的并发任务。
-
drainCurrentTickTasks 的非阻塞读取：
- 这是实现“等待所有当前tick任务完成”的关键。它使用 for 循环结合 select 和 default 语句。
- select { case f :=
- 每次启动并发任务时，sync.WaitGroup 的计数器会增加（el.wg.Add(1)）。任务完成后，计数器减少（el.wg.Done()）。
- 在 drain 循环结束后，调用 el.wg.Wait() 会阻塞，直到所有通过 curFunc 启动的并发任务都执行完毕。这确保了在一个“tick”内的所有并发任务完全结束后，eventLoopRunner 才会继续处理下一个 nextFunc 任务。
-
优雅退出：
- Quit() 方法通过关闭 quit 通道来向 eventLoopRunner 发送退出信号。
- eventLoopRunner 收到 quit 信号后，会尝试清空并执行 nextFunc 和 curFunc 中剩余的任务，然后安全退出。
- 关闭 nextFunc 和 curFunc 通道也很重要，它会使得 eventLoopRunner 中的 f, ok :=
-
同步与竞态条件：
- sync.WaitGroup 是确保所有并发任务完成的关键。它避免了手动计数器和互斥锁带来的复杂性和潜在的竞态条件。
- select 语句本身是原子操作，确保了通道操作的安全性。
总结
通过巧妙地利用Go语言的通道和 sync.WaitGroup，我们可以构建一个高效、低延迟且并发安全的事件循环。这种设计避免了传统方法中CPU空转和高延迟的问题，同时提供了清晰的任务分类和调度机制。在实际应用中，可以根据具体需求调整通道容量，并考虑更复杂的错误处理和资源清理逻辑。理解并掌握这种基于通道的并发模式，对于编写高性能的Go并发程序至关重要。
