Go语言任务调度

Goroutine简介定义:Goroutine是一种轻量级线程,由Go运行时管理。它比操作系统级别的线程更轻量,创建成本低,切换速度快。创建方式:通过go关键字启动一个新goroutine。gofunc(){fmt.Println("Hellofromgoroutine"

Goroutine简介

  • 定义:Goroutine是一种轻量级线程,由Go运行时管理。它比操作系统级别的线程更轻量,创建成本低,切换速度快。
  • 创建方式:通过go关键字启动一个新goroutine。
go func() {
    fmt.Println("Hello from goroutine")
}()

Channel的作用

  • 定义:Channel是goroutine之间通信的管道,用于发送和接收数据。
  • 声明与使用
ch := make(chan int) // 声明一个整型channel
ch <- 42             // 向channel发送数据
x := <-ch            // 从channel接收数据
fmt.Println(x)       // 输出: 42

同步与异步

  • 同步:等待一个操作完成后再执行下一步。
  • 异步:不等待操作完成,直接执行下一步。

并发模式

生产者消费者模型:通过channel连接生产者和消费者。

func producer(ch chan<- int) {
    for i := 0; i < 10; i++ {
        ch <- i
    }
    close(ch)
}

func consumer(ch <-chan int) {
    for n := range ch {
        fmt.Println(n)
    }
}

func main() {
    ch := make(chan int)
    go producer(ch)
    consumer(ch)
}

工作池:通过固定数量的goroutine处理任务队列。

type WorkerPool struct {
    jobs  chan int
    done  chan bool
    count int
}

func NewWorkerPool(n int) *WorkerPool {
    return &WorkerPool{
        jobs:  make(chan int, 100),
        done:  make(chan bool, n),
        count: n,
    }
}

func (p *WorkerPool) Start() {
    for i := 0; i < p.count; i++ {
        go func() {
            for job := range p.jobs {
                fmt.Println("Working on", job)
                time.Sleep(time.Second)
            }
            p.done <- true
        }()
    }
}

func (p *WorkerPool) AddJob(jobs ...int) {
    for _, job := range jobs {
        p.jobs <- job
    }
}

func (p *WorkerPool) Wait() {
    for i := 0; i < p.count; i++ {
        <-p.done
    }
    close(p.jobs)
}

func main() {
    wp := NewWorkerPool(5)
    wp.Start()
    wp.AddJob(1, 2, 3, 4, 5)
    wp.Wait()
}

超时控制

使用context包可以方便地设置超时。

ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()

select {
case <-ctx.Done():
    fmt.Println("Timed out")
case result := <-ch:
    fmt.Println("Result:", result)
}

错误处理

在并发程序中,错误处理尤为重要。

func doWork(ch chan<- error) {
    defer func() {
        if r := recover(); r != nil {
            ch <- errors.New("panic occurred")
        }
    }()

    // 可能引发panic的操作
    fmt.Println(1 / 0)
}

func main() {
    ch := make(chan error)
    go doWork(ch)
    if err := <-ch; err != nil {
        log.Fatal(err)
    }
}

性能考量

  • Goroutine的数量不宜过多,否则会导致上下文切换频繁,影响性能。
  • Channel的缓冲区大小需合理设置,过大会占用内存,过小则可能造成goroutine阻塞。

实战案例

假设我们需要开发一个简单的Web服务器,该服务器能够同时处理多个客户端请求,并且每个请求都可能触发一些耗时较长的操作(如数据库查询)。我们可以利用goroutine和channel来实现这一功能:

package main

import (
    "fmt"
    "log"
    "net/http"
)

func handleRequest(w http.ResponseWriter, r *http.Request) {
    // 模拟耗时操作
    ch := make(chan string)
    go func() {
        time.Sleep(2 * time.Second)
        ch <- "Hello, World!"
    }()

    // 等待结果返回
    result := <-ch
    fmt.Fprintln(w, result)
}

func main() {
    http.HandleFunc("/", handleRequest)
    log.Fatal(http.ListenAndServe(":8080", nil))
}

Select语句

Select语句:类似于多路复用器,可以在多个channel操作中选择一个准备好的操作执行。

ch1 := make(chan int)
ch2 := make(chan int)

go func() {
    time.Sleep(1 * time.Second)
    ch1 <- 1
}()

go func() {
    time.Sleep(2 * time.Second)
    ch2 <- 2
}()

select {
case v := <-ch1:
    fmt.Println("Received from ch1:", v)
case v := <-ch2:
    fmt.Println("Received from ch2:", v)
}
  • Default case:当没有channel准备好时,默认分支会执行。
ch := make(chan int)

select {
case v := <-ch:
    fmt.Println("Received:", v)
default:
    fmt.Println("No data received yet")
}

Context包

Context包:用于传递取消信号、截止时间和其他请求范围的数据。

package main

import (
    "context"
    "fmt"
    "log"
    "time"
)

func doWork(ctx context.Context, ch chan<- string) {
    select {
    case <-ctx.Done():
        ch <- "Operation canceled"
        return
    case <-time.After(2 * time.Second):
        ch <- "Work completed"
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    ch := make(chan string)

    go doWork(ctx, ch)

    time.Sleep(1 * time.Second)
    cancel() // 取消操作

    result := <-ch
    fmt.Println(result)
}

工作窃取(Work Stealing)

工作窃取:一种优化策略,允许空闲的goroutine从其他繁忙的goroutine那里“窃取”任务。

package main

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

var wg sync.WaitGroup

func worker(id int, tasks chan int, results chan<- int) {
    for task := range tasks {
        fmt.Printf("Worker %d processing task %d\n", id, task)
        time.Sleep(1 * time.Second)
        results <- task * task
    }
    wg.Done()
}

func main() {
    numWorkers := 5
    numTasks := 10

    tasks := make(chan int, numTasks)
    results := make(chan int, numTasks)

    wg.Add(numWorkers)
    for i := 0; i < numWorkers; i++ {
        go worker(i, tasks, results)
    }

    for i := 1; i <= numTasks; i++ {
        tasks <- i
    }
    close(tasks)

    wg.Wait()

    for i := 0; i < numTasks; i++ {
        fmt.Println(<-results)
    }
}

闭包与匿名函数

闭包:允许goroutine访问外部变量。

package main

import (
    "fmt"
    "time"
)

func main() {
    values := []int{1, 2, 3, 4, 5}

    for _, value := range values {
        go func(v int) {
            time.Sleep(1 * time.Second)
            fmt.Println(v)
        }(value)
    }

    time.Sleep(2 * time.Second) // 确保所有goroutine完成
}

通道容量与同步

通道容量:带缓冲的channel可以减少goroutine之间的同步开销。

ch := make(chan int, 2) // 容量为2的channel

ch <- 1
ch <- 2

fmt.Println(<-ch)
fmt.Println(<-ch)

同步问题:如果多个goroutine尝试同时访问共享资源,可能会导致竞态条件。

var count int
var mutex sync.Mutex

func increment() {
    mutex.Lock()
    count++
    mutex.Unlock()
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            increment()
        }()
    }

    wg.Wait()
    fmt.Println(count) // 应输出1000
}

并发安全的数据结构

并发安全的容器:Go标准库提供了多种并发安全的数据结构。

package main

import (
    "container/list"
    "fmt"
    "sync"
)

type SafeList struct {
    list *list.List
    lock sync.Mutex
}

func (s *SafeList) PushFront(v interface{}) {
    s.lock.Lock()
    defer s.lock.Unlock()
    s.list.PushFront(v)
}

func (s *SafeList) Front() interface{} {
    s.lock.Lock()
    defer s.lock.Unlock()
    if s.list.Len() == 0 {
        return nil
    }
    return s.list.Front().Value
}

func main() {
    l := &SafeList{list: list.New()}
    var wg sync.WaitGroup

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            l.PushFront(i)
        }(i)
    }

    wg.Wait()
    fmt.Println(l.Front()) // 应输出9
}
点赞 0
收藏 0
分享
本文参与登链社区写作激励计划 ,好文好收益,欢迎正在阅读的你也加入。

0 条评论

请先 登录 后评论
天涯学馆
天涯学馆
0x9d6d...50d5
资深大厂程序员,12年开发经验,致力于探索前沿技术!