如何在 Golang 中构建多线程任务分发系统_Golang 并发调度框架项目示例

type Task struct {
    ID   int
    Data string
}
<p>func worker(id int, jobs <-chan Task, results chan<- error) {
for task := range jobs {
// 模拟任务处理
fmt.Printf("Worker %d processing task %d: %s\n", id, task.ID, task.Data)
time.Sleep(time.Second) // 模拟耗时操作
results <- nil // 表示成功
}
}</p>

func main() {
    const numWorkers = 3
    const numTasks = 5
<pre class='brush:php;toolbar:false;'>jobs := make(chan Task, numTasks)
results := make(chan error, numTasks)

// 启动 worker
for w := 1; w <= numWorkers; w++ {
    go worker(w, jobs, results)
}

// 发送任务
for t := 1; t <= numTasks; t++ {
    jobs <- Task{ID: t, Data: fmt.Sprintf("data-%d", t)}
}
close(jobs)

// 收集结果
for a := 1; a <= numTasks; a++ {
    <-results
}

fmt.Println("All tasks completed.")

type Scheduler struct {
    workers   int
    jobs      chan Task
    results   chan error
    stopChan  chan struct{}
}
<p>func NewScheduler(workers, queueSize int) *Scheduler {
return &Scheduler{
workers:  workers,
jobs:     make(chan Task, queueSize),
results:  make(chan error, queueSize),
stopChan: make(chan struct{}),
}
}</p><p>func (s *Scheduler) Start() {
for i := 0; i < s.workers; i++ {
go func(id int) {
for {
select {
case task, ok := <-s.jobs:
if !ok {
return
}
fmt.Printf("Worker %d processing %d\n", id, task.ID)
time.Sleep(time.Second)
s.results <- nil
case <-s.stopChan:
return
}
}
}(i + 1)
}
}</p><p>func (s *Scheduler) Submit(task Task) {
s.jobs <- task
}</p><p>func (s *Scheduler) Close() {
close(s.jobs)
close(s.results)
close(s.stopChan)
}</p>