Goroutine
Goroutine 是 Go 的轻量级线程，由 Go 运行时管理。创建 Goroutine
// 使用 go 关键字启动
func sayHello() {
fmt.Println("Hello")
}
go sayHello() // 异步执行
// 匿名函数
go func() {
fmt.Println("匿名函数")
}()
// 带参数
go func(msg string) {
fmt.Println(msg)
}("hello")
Goroutine 特点
| 特点 | 说明 |
|---|---|
| 轻量级 | 初始栈只有 2KB，可动态增长 |
| 数量大 | 可以创建数十万个 Goroutine |
| 调度高效 | Go 运行时调度，无需系统调用 |
| 共享内存 | 同一进程内，可直接访问共享变量 |
等待 Goroutine 完成
import "sync"
func main() {
var wg sync.WaitGroup
for i := 0; i < 5; i++ {
wg.Add(1) // 计数 +1
go func(n int) {
defer wg.Done() // 计数 -1
fmt.Println(n)
}(i)
}
wg.Wait() // 等待所有 Goroutine 完成
fmt.Println("全部完成")
}
Channel
Channel 是 Goroutine 之间通信的管道。创建 Channel
// 无缓冲 Channel
ch := make(chan int)
// 有缓冲 Channel
ch := make(chan int, 10) // 容量为 10
// 只发送 Channel
var sendOnly chan<- int
// 只接收 Channel
var recvOnly <-chan int
发送和接收
ch := make(chan int)
// 发送
go func() {
ch <- 42 // 发送数据
}()
// 接收
value := <-ch // 接收数据
fmt.Println(value)
// 接收并丢弃
<-ch
无缓冲 vs 有缓冲
// 无缓冲：同步通信，发送和接收必须同时就绪
ch := make(chan int)
go func() {
ch <- 1 // 阻塞，直到有接收者
}()
<-ch // 阻塞，直到有发送者
// 有缓冲：异步通信，缓冲满时发送阻塞，空时接收阻塞
ch := make(chan int, 3)
ch <- 1 // 不阻塞
ch <- 2 // 不阻塞
ch <- 3 // 不阻塞
ch <- 4 // 阻塞，缓冲已满
关闭 Channel
ch := make(chan int, 3)
ch <- 1
ch <- 2
close(ch) // 关闭 channel
// 关闭后仍可接收
fmt.Println(<-ch) // 1
fmt.Println(<-ch) // 2
fmt.Println(<-ch) // 0（默认值，channel 已空）
// 检查 channel 是否关闭
value, ok := <-ch
if !ok {
fmt.Println("channel 已关闭")
}
// 向已关闭的 channel 发送会 panic
// ch <- 3 // panic!
遍历 Channel
ch := make(chan int, 5)
go func() {
for i := 0; i < 5; i++ {
ch <- i
}
close(ch) // 必须关闭，否则 range 会阻塞
}()
// 使用 range 遍历
for value := range ch {
fmt.Println(value)
}
select
select 用于同时监听多个 Channel。ch1 := make(chan int)
ch2 := make(chan string)
go func() {
time.Sleep(time.Second)
ch1 <- 1
}()
go func() {
time.Sleep(2 * time.Second)
ch2 <- "hello"
}()
// 等待多个 channel
select {
case v := <-ch1:
fmt.Println("ch1:", v)
case v := <-ch2:
fmt.Println("ch2:", v)
case <-time.After(3 * time.Second):
fmt.Println("超时")
}
非阻塞操作
select {
case v := <-ch:
fmt.Println("接收:", v)
default:
fmt.Println("没有数据")
}
超时控制
select {
case result := <-ch:
fmt.Println("结果:", result)
case <-time.After(5 * time.Second):
fmt.Println("超时")
}
并发模式
生产者-消费者
func producer(ch chan<- int) {
for i := 0; i < 10; i++ {
ch <- i
fmt.Println("生产:", i)
}
close(ch)
}
func consumer(ch <-chan int) {
for v := range ch {
fmt.Println("消费:", v)
}
}
func main() {
ch := make(chan int, 5)
go producer(ch)
consumer(ch)
}
工作池
func worker(id int, jobs <-chan int, results chan<- int) {
for job := range jobs {
fmt.Printf("Worker %d 处理 Job %d\n", id, job)
time.Sleep(time.Second)
results <- job * 2
}
}
func main() {
jobs := make(chan int, 100)
results := make(chan int, 100)
// 启动 3 个 worker
for w := 1; w <= 3; w++ {
go worker(w, jobs, results)
}
// 发送 9 个任务
for j := 1; j <= 9; j++ {
jobs <- j
}
close(jobs)
// 收集结果
for i := 1; i <= 9; i++ {
<-results
}
}
扇入扇出
// 扇出：一个 channel 分发给多个 goroutine
func fanOut(ch <-chan int, n int) []<-chan int {
outs := make([]<-chan int, n)
for i := 0; i < n; i++ {
out := make(chan int)
outs[i] = out
go func() {
for v := range ch {
out <- v * 2
}
close(out)
}()
}
return outs
}
// 扇入：多个 channel 合并为一个
func fanIn(chs ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
for _, ch := range chs {
wg.Add(1)
go func(c <-chan int) {
defer wg.Done()
for v := range c {
out <- v
}
}(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
同步原语
sync.Mutex
type Counter struct {
mu sync.Mutex
value int
}
func (c *Counter) Inc() {
c.mu.Lock()
defer c.mu.Unlock()
c.value++
}
func (c *Counter) Value() int {
c.mu.Lock()
defer c.mu.Unlock()
return c.value
}
sync.RWMutex
type Cache struct {
mu sync.RWMutex
data map[string]string
}
func (c *Cache) Get(key string) string {
c.mu.RLock()
defer c.mu.RUnlock()
return c.data[key]
}
func (c *Cache) Set(key, value string) {
c.mu.Lock()
defer c.mu.Unlock()
c.data[key] = value
}
sync.Once
var once sync.Once
var instance *Singleton
func GetInstance() *Singleton {
once.Do(func() {
instance = &Singleton{}
})
return instance
}
sync.Pool
var bufferPool = sync.Pool{
New: func() interface{} {
return new(bytes.Buffer)
},
}
func process() {
buf := bufferPool.Get().(*bytes.Buffer)
defer func() {
buf.Reset()
bufferPool.Put(buf)
}()
// 使用 buf...
}
Context
Context 用于控制 Goroutine 的生命周期。import "context"
func worker(ctx context.Context) {
for {
select {
case <-ctx.Done():
fmt.Println("收到取消信号")
return
default:
fmt.Println("工作中...")
time.Sleep(time.Second)
}
}
}
func main() {
// 可取消的 context
ctx, cancel := context.WithCancel(context.Background())
go worker(ctx)
time.Sleep(3 * time.Second)
cancel() // 取消
time.Sleep(time.Second)
}
// 带超时
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
// 带截止时间
ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second))
defer cancel()
// 带值
ctx := context.WithValue(context.Background(), "key", "value")
value := ctx.Value("key").(string)