并发模型（CSP理论、Goroutine轻量级线程）

1. 语法讲解

并发与并行概念：

并发：多个任务在同一时间段内交替执行（看起来同时）【单cpu】
并行：多个任务在同一时刻真正同时执行【多核】

CSP理论（Communicating Sequential Processes）：

核心思想：”不要通过共享内存来通信，而应该通过通信来共享内存”
组成要素：Process（进程/协程） + Channel（通道）

func limitedWorkerPool(task []string, max int) {
	start := time.Now()

	semaphore := make(chan struct{}, max)
	var wg sync.WaitGroup
	for i, task := range task {
		wg.Add(1)
		fmt.Println("go before", task, "index", i)
		go func(task string, i int) { // go 决定当前匿名函数并行执行
			semaphore <- struct{}{} // 当semaphore 满了，即已经有max 个任务在执行，代码停在这里，直达有空位
			defer func() {
				<-semaphore
				wg.Done()
			}()
			fmt.Println("go inner", i, "start")
			fetchData(task, nil)
		}(task, i)
	}
	wg.Wait() // 等待所有任务完成才执行后面的代码
	fmt.Println("All tasks completed 3", time.Since(start))
}

这段代码是一个经典的 使用带缓冲 Channel 实现并发限制（信号量）的工作池（Worker Pool） 模式。

结果打印顺序每次不一样、fetchData 执行顺序乱，是完全正常且符合预期的。

一、代码逻辑分析

semaphore := make(chan struct{}, max)：
创建了一个带缓冲的 channel，容量为 max。这充当了信号量（Semaphore），用于限制同时运行的 Goroutine 数量不超过 max。
for i, task := range task 循环：
遍历任务列表。每遍历到一个任务，就 wg.Add(1) 并开启一个新的 Goroutine。
Goroutine 内部逻辑：
- semaphore <- struct{}{}：向 channel 发送一个空结构体。如果 channel 已满（即当前正在运行的任务数达到了 max），这个操作会阻塞，直到有别的 Goroutine 执行 <-semaphore 释放出位置。
- defer func() { <-semaphore; wg.Done() }()：在函数退出时，从 channel 接收一个值（释放信号量），并调用 wg.Done() 通知 WaitGroup 任务完成。
- fmt.Println("go inner", i, "start")：打印内部启动日志。
- fetchData(task, nil)：执行实际的任务（比如发起 HTTP 请求）。

二、为什么打印结果每次不一样，且执行顺序乱？

这是由 Go 语言的并发调度机制（Goroutine Scheduler） 和 操作系统线程调度 决定的。

1. 主协程与子协程的抢占式调度

go func(...) 只是声明要启动一个协程，它并不会立刻执行完毕。
主协程（main 或调用 limitedWorkerPool 的协程）在 for 循环中启动协程的速度非常快。当主协程启动完第 0、1、2 个协程后，可能 CPU 时间片才轮到这些子协程开始运行。
即使启动了 5 个协程，它们谁先抢到 CPU 资源，谁就先执行 fmt.Println("go inner...") 和 fetchData。 这个顺序是由 Go 运行时和操作系统底层的线程调度器决定的，每次运行时的系统负载、内存状态都不同，所以顺序是随机的。

2. `semaphore` 的阻塞机制

假设 max = 2（只允许 2 个并发）：

前 2 个抢到 semaphore 的协程会继续执行 fetchData。
第 3、4、5 个协程会阻塞在 semaphore <- struct{}{} 这一行，等待前 2 个执行完并释放信号量。
当第 1 个任务完成时，它会释放信号量，此时在阻塞队列中的协程不一定是按先进先出（FIFO）的顺序被唤醒的。Go 的 channel 唤醒机制在并发场景下也是非确定性的，谁抢到算谁的。所以 go inner 打印的顺序和 fetchData 执行的顺序完全无法预测。

3. `fetchData` 内部的耗时不确定

从输出可以看到：

go inner 4 start
go inner 1 start
go inner 0 start
Fetcing data from https://www.google.com delay 100 ms
go inner 2 start
Fetcing data from https://www.baidu.com delay 100 ms
go inner 3 start
Fetcing data from https://www.yandex.com delay 100 ms
Fetcing data from https://www.yahoo.com delay 100 ms
Fetcing data from https://www.bing.com delay 100 ms

可以看到：

go inner 4 比 go inner 1 先打印，说明 4 号协程先抢到了 CPU。
go inner 0 启动后，紧接着开始 Fetching data from google，说明 0 号协程执行很快。
而 Fetching data from baidu 在 go inner 2 之后才打印，说明网络请求（或模拟延迟）的耗时各不相同。网络 I/O 操作是非常不可控的，这会导致即使并发启动，各任务实际完成的绝对时间也有很大差异。

如果并发执行，但要求结果按顺序收集，那么可以使用以下模式：

使用 Channel 收集结果：创建一个结果 channel，每个协程执行完把结果发到 channel，主协程按顺序从 channel 读取，但读取顺序依然取决于谁先写入。
使用索引数组：创建一个和 task 长度一样的 result 切片，协程执行完后将结果写入 result[i]，最后 wg.Wait() 结束后统一按顺序遍历输出。这是最常用的实现“并发执行、有序输出”的方法。

// 显示当前Goroutine数量和CPU核心数，设置最大并行度为4
func showGoRoutine() {
	fmt.Printf("当前Goroutine数量: %d\n", runtime.NumGoroutine())
	fmt.Printf("CPU核心数: %d\n", runtime.NumCPU())

	// 设置最大并行度
	runtime.GOMAXPROCS(4)
}


func worker(context context.Context, id int) {
	for {
		select {
		case <-context.Done():
			fmt.Println("Worker", id, "DONE")
			return
		case <-time.After(time.Second * 1): // time1
			fmt.Println("Worker", id, "working")
		}
	}
}

ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) // time2
defer cancel()
go worker(ctx, 1)
go worker(ctx, 2)

time.Sleep(time.Second * 3) // time3

上面这段代码结束时间一次取决于 time3> time2 >time1

time1 决定每隔几秒进行一次打印, 因此当time2, time3 都比time1 大时，会进行连续打印

time3 决定使用当前ctx 的程序整体运行时间，如果time1 大于time3, 等time3 时间耗完会直接ctx.done

time2 决定整个程序退出时间
如果time3 大于time2， time2时间优先到达后，defer cancel() 会直接执行，关掉time3 覆盖的程序，不会再打印 worker working
如果time2 大于time3， time3时间优先到达后，会直接ctx.done, time2 时间到达后再执行 defer cancel()

2. 应用场景

Goroutine适用场景：

高并发Web服务器：每个请求一个Goroutine
数据处理流水线：多个处理阶段并行执行
实时消息推送：WebSocket连接管理
定时任务调度：后台定时执行任务
并发爬虫：同时抓取多个网页

传统多线程 vs Goroutine：

特性	传统线程	Goroutine
创建成本	1-2MB	2KB
创建速度	慢	快
调度方式	操作系统调度	Go运行时调度
上下文切换	完整线程切换	用户态轻量切换

3. 总结

并发vs并行：并发是交替执行，并行是真正同时执行
CSP理论：通过通信共享内存，而非通过共享内存通信
Goroutine优势：
- 轻量级（2KB起步）
- 创建快速
- 调度高效（用户态调度）
- 内存占用少
最佳实践：
- 使用sync.WaitGroup等待Goroutine完成
- 使用context控制Goroutine生命周期
- 合理控制并发数量
- 注意Goroutine泄漏问题（常见线路问题，线路已经阻塞，还在不停创建）
适用场景：高并发服务、数据处理流水线、实时系统等

练习

Channel（管道）

1. 语法讲解

Channel是Go语言中各个并发执行体间的通信机制，是类型相关的管道，用于在goroutine之间传递数据和同步执行。

不要通过共享内存来通信，而应通过通信来共享内存

// 创建channel
ch1 := make(chan int)        // 无缓冲channel
ch2 := make(chan int, 10)    // 有缓冲channel，容量10

// 操作
ch1 <- 42      // 发送数据到channel
value := <-ch1 // 从channel接收数据
close(ch1)     // 关闭channel

// 特殊用法
value, ok := <-ch1 // 检查channel是否关闭

缓冲区区别：

无缓冲channel：同步通信，发送和接收必须同时准备好
有缓冲channel：异步通信，缓冲区满时发送阻塞，空时接收阻塞

select多路复用：同时等待多个channel操作。

2. 应用场景

goroutine通信：协程间传递数据
任务分发：生产者-消费者模式
结果收集：多个并行任务的结果聚合
同步控制：替代锁机制，协调执行顺序
超时控制：配合定时器实现超时机制


// 生产者-消费者模式

func CreateDeploySheet(DeploySheetChan chan *DeploySheet, count int) {
	for i := 0; i < count; i++ {
		DeploySheetChan <- &DeploySheet{
			ProjectName:    "project" + strconv.Itoa(i),
			Environment:    "env" + strconv.Itoa(i),
			ReleaseVersion: "release" + strconv.Itoa(i),
			Status:         "pending",
		}
		fmt.Printf("CreateDeploySheet %d success\n", i)
		time.Sleep(time.Second * 1)
	}
	close(DeploySheetChan)
}

func AuditDeploySheet(deploySheetChan chan *DeploySheet, resultChan chan *DeploySheet, wg *sync.WaitGroup) {
	defer wg.Done()
	for deploySheet := range deploySheetChan {
		fmt.Printf("AuditDeploySheet %s %s %s %s\n", deploySheet.ProjectName, deploySheet.Environment, deploySheet.ReleaseVersion, deploySheet.Status)
		deploySheet.Status = "AuditFinished"
		time.Sleep(time.Second * 1)
		resultChan <- deploySheet
	}
}

func FinishDeploySheet(resultChan chan *DeploySheet, done chan bool) {
	count := 0
	for deploySheet := range resultChan {
		fmt.Printf("FinishDeploySheet %s %s %s %s\n", deploySheet.ProjectName, deploySheet.Environment, deploySheet.ReleaseVersion, deploySheet.Status)
		count++
		time.Sleep(time.Second * 1)
	}

	done <- true
}

DeploySheetChan := make(chan *DeploySheet)
ResultChan := make(chan *DeploySheet)
done := make(chan bool)

go CreateDeploySheet(DeploySheetChan, 3)
// 借助 sync.WaitGroup 等待所有审核 worker 退出后，再关闭 ResultChan
var wg sync.WaitGroup
for i := 0; i < 3; i++ {
    wg.Add(1)
    go AuditDeploySheet(DeploySheetChan, ResultChan, &wg)
}
go func() {
    wg.Wait()
    close(ResultChan)
}()
go FinishDeploySheet(ResultChan, done)

<-done // 通道内数据只接收，不保留,等待所有处理完成

fmt.Println("hello world", done)

下是详细的执行过程拆解：

1. `go` 关键字的作用：发射后不管

当代码执行到 go AuditDeploySheet(...) 时：

Go 运行时会创建一个新的协程。
把它加入到调度队列中准备运行。
主协程（当前的代码执行流）不会停下来等它，而是立刻继续往下执行 for 循环的下一次迭代，直到循环结束。

同理，当代码执行到 go FinishDeploySheet(ResultChan, done) 时：

主协程立刻创建一个新的协程去跑 FinishDeploySheet。
主协程自己继续往下执行（如果有后续代码的话），或者如果主协程（比如 main 函数）没有阻塞，直接退出了，整个程序就结束了。

2. 这两个协程是“同时”运行的

在你的代码中，AuditDeploySheet（生产者）和 FinishDeploySheet（消费者）是并发运行的。它们互不干涉对方的启动。

我们可以想象这样一个场景（假设这是两条时间线）：

时间线 A (生产者 Worker)：

go AuditDeploySheet 启动。
它发现 DeploySheetChan 里暂时没数据，于是阻塞（挂起），等待数据。
（等待中…）
数据来了，它开始处理数据。
处理完一条，把结果塞进 ResultChan，再去拿一下条。

时间线 B (消费者 Consumer)：

go FinishDeploySheet 启动（它通常在 Worker 启动后极短时间内就启动了）。
它执行类似 for res := range ResultChan 的逻辑。
它发现 ResultChan 里暂时没有数据（因为 Worker 还没处理完），于是阻塞（挂起），等待数据。
（等待中…）
突然 ResultChan 里有数据了（因为时间线 A 的 Worker 塞进去了），它立刻被唤醒，拿走数据，执行后续逻辑。
拿完后又阻塞，等待下一条。

3. 核心机制：Channel 的阻塞与唤醒

你可能会问：“如果消费者一开始就执行了，但通道里没数据，它不就跑完退出了吗？”

不会的。 这依赖于 Go 语言中 Channel（通道）的阻塞特性：

当消费者尝试从空的 Channel 读取数据时，它会阻塞（暂停），直到 Channel 里有数据或者 Channel 被关闭。
当生产者尝试向满的 Channel（或无缓冲 Channel）写入数据时，它也会阻塞，直到有人把数据取走。

所以，实际上这两个协程的节奏是这样配合的：

消费者先跑起来，发现没数据，就在 ResultChan 那里“睡”着了。
生产者跑起来，往 ResultChan 里塞了一个数据。
操作系统/Go 调度器发现消费者在等这个 Channel，立刻把消费者“叫醒”。
消费者拿走数据去处理，生产者继续塞下一条。
直到最后，生产者全部干完，close(ResultChan)。
消费者读到关闭信号，知道没活了，结束 for 循环，退出。

4. 一个至关重要的前提条件（必须注意）

虽然 go FinishDeploySheet 会立刻执行，但它绝对不能比 close(ResultChan) 晚执行，否则它会永远卡死（死锁）。

这段代码：

go func() {
    wg.Wait()
    close(ResultChan)
}()
go FinishDeploySheet(ResultChan, done)

因为有了 wg.Wait() 这个“监工”，close(ResultChan) 必然发生在所有 AuditDeploySheet 执行完毕之后。
只要 FinishDeploySheet 是在 close(ResultChan) 之前启动的（在这段代码里它确实是），即使它一开始因为没数据而阻塞，它也绝对能等到所有数据被写入、并且 Channel 被关闭的那一刻，从而正常退出。

总结

AuditDeploySheet(DeploySheetChan, ResultChan, &wg) 执行过程中，FinishDeploySheet(ResultChan, done)“会执行。”
并且它通常会在 AuditDeploySheet 还没有处理完所有任务时，就已经开始运行并在通道上等待数据了。它们是通过 Channel 的信号机制 来实现相互协作（同步）的，而不是通过先后启动顺序来决定的。

其他使用场景


func main() {
    // 1. select多路复用
    ch1 := make(chan string)
    ch2 := make(chan string)
    
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "来自ch1"
    }()
    
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "来自ch2"
    }()
    
    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-ch1:
            fmt.Println("收到:", msg1)
        case msg2 := <-ch2:
            fmt.Println("收到:", msg2)
        case <-time.After(3 * time.Second): // 超时控制
            fmt.Println("超时!")
            return
        }
    }
    
    // 2. 定时器与Ticker
    ticker := time.NewTicker(500 * time.Millisecond)
    done := make(chan bool)
    
    go func() {
        for {
            select {
            case <-done:
                return
            case t := <-ticker.C:
                fmt.Println("定时触发 at", t.Format("15:04:05"))
            }
        }
    }()
    
    time.Sleep(2 * time.Second)
    ticker.Stop()
    done <- true
    fmt.Println("Ticker停止")
    
    // 3. 工作池模式
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    
    // 启动3个worker
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }
    
    // 发送5个任务
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)
    
    // 收集结果
    for i:=0; i<=5; i++ {
		value := <-results
        fmt.Printf("Worker 处理结果 %d\n", value)
    }
}

func worker(id int, jobs <-chan int, results chan<- int) {
    for j := range jobs {
        fmt.Printf("Worker %d 处理任务 %d\n", id, j)
        time.Sleep(time.Second)
        results <- j * 2
    }
}

3. 总结

Channel类型：无缓冲(同步)/有缓冲(异步)
通信机制：goroutine间安全传递数据，解决并发访问问题
select多路复用：同时处理多个channel操作
模式应用：生产者-消费者、工作池、结果聚合等并发模式

练习

同步原语（sync.Mutex互斥锁、sync.WaitGroup等待组、sync.RWMutex读写锁）

1. 语法讲解

为什么需要同步原语？
在并发编程中，多个goroutine同时访问共享资源时会产生竞态条件，导致数据不一致。

sync.Mutex（互斥锁）

保证同一时间只有一个goroutine能访问共享资源
两个方法：Lock() 和 Unlock()
使用后必须释放，否则会导致死锁

sync.RWMutex（读写锁）

允许多个读操作或一个写操作
读锁：RLock() / RUnlock() （共享锁）
写锁：Lock() / Unlock() （互斥锁）
适合”读多写少”的场景

sync.WaitGroup（等待组）

等待一组goroutine完成工作
三个方法：Add()、Done()、Wait()
用于协调多个goroutine的执行顺序

type Stock struct {
	Stock   int
	rwMutex sync.RWMutex
}

func (s *Stock) GetStock() int {
	s.rwMutex.RLock()
	defer func() {
		s.rwMutex.RUnlock()
	}()
	time.Sleep(time.Millisecond * 100)
	stock := s.Stock
	return stock
}

func (s *Stock) IncreaseStock(num int) int {
	s.rwMutex.Lock()
	s.Stock += num
	time.Sleep(time.Millisecond * 100)
	s.rwMutex.Unlock()
	fmt.Println("增加库存成功，当前库存为：", s.Stock)
	return s.Stock
}

func (s *Stock) DecreaseStock(num int) int {
	s.rwMutex.Lock()
	defer func() {
		s.rwMutex.Unlock()
	}()
	if s.Stock < num {
		fmt.Println("库存不足")
		return s.Stock
	}
	s.Stock -= num
	time.Sleep(time.Millisecond * 100)
	fmt.Println("减少库存成功，当前库存为：", s.Stock)
	return s.Stock
}

type Counter struct {
	Count int32
}

func (c *Counter) Inc() {
	time.Sleep(time.Millisecond * 100)
	atomic.AddInt32(&c.Count, 1)
	fmt.Println("增加计数成功，当前计数为：", c.Count)
}

func (c *Counter) Dec() {
	time.Sleep(time.Millisecond * 100)
	atomic.AddInt32(&c.Count, -1)
	fmt.Println("减少计数成功，当前计数为：", c.Count)
}
func (c *Counter) GetCount() int32 {
	return atomic.LoadInt32(&c.Count)
}

stock := &Stock{
		Stock: 100,
	}
	fmt.Println("当前库存为：", stock.GetStock())

	start := time.Now()
	var wg sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			stock.DecreaseStock(1)
		}(i)
	}

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			currentStock := stock.GetStock()
			fmt.Println("当前库存为：", currentStock)
		}(i)
	}
	wg.Wait()
	fmt.Println("耗时：", time.Since(start)) // 耗时： 10.1979415s

	start1 := time.Now()
	counter := &Counter{
		Count: 100,
	}
	var wg2 sync.WaitGroup
	// for i := 0; i < 100; i++ {
	// 	wg2.Add(1)
	// 	go func(i int) {
	// 		defer wg2.Done()
	// 		counter.Inc()
	// 	}(i)
	// }

	DecFunc := func() {
		defer wg2.Done()
		counter.Dec()
	}

	for i := 0; i < 100; i++ {
		wg2.Add(1)
		go DecFunc()
	}

	for i := 0; i < 10; i++ {
		wg2.Add(1)
		go func() {
			defer wg2.Done()
			count := counter.GetCount()
			fmt.Println("当前计数为：", count)
		}()
	}
	wg2.Wait()
	fmt.Println("耗时：", time.Since(start1)) //  101.235875ms

2. 应用场景

Mutex适用场景：

银行账户余额更新（防止超额取款）
电商库存扣减（防止超卖）
计数器累加操作

WaitGroup适用场景：

批量文件下载，等待所有下载完成
并行数据处理，等待所有处理结果
服务启动时等待多个初始化任务完成

RWMutex适用场景：

配置信息读取（频繁读，偶尔更新）
缓存系统（大量查询，少量更新）
热点数据访问（如商品详情页）

3. 总结

Mutex互斥锁：保护临界区，同一时间只允许一个goroutine访问
RWMutex读写锁：读多写少场景下的性能优化
WaitGroup等待组：协调多个goroutine的完成状态
使用原则：
- 锁的粒度要适中，不要过大或过小，最小细粒度到一个字段
- 使用defer确保锁的释放
- 避免在持有锁时进行耗时操作
- 读写分离场景优先使用RWMutex
- 简单场景优先使用原子操作【atomic 操作效率远高于锁】

练习

并发模式（生产者-消费者、扇入/扇出、Pipeline）

1. 语法讲解

生产者-消费者模式

生产者-消费者模式是并发编程中最经典的模式之一，解决生产者和消费者速度不匹配的问题。
就像食堂阿姨打饭和同学们吃饭，赚钱和花钱的关系是一样的。

扇入/扇出模式：

扇出：一个channel分发给多个goroutine处理（一产多消）
扇入：多个channel合并到一个channel（多产一消）

Pipeline模式：将复杂任务分解为多个处理阶段，每个阶段通过channel连接，形成处理流水线。

2. 应用场景

生产者-消费者适用场景：

消息队列系统（如订单处理、日志收集）
文件上传后的异步处理
爬虫系统的URL调度与页面解析

扇入/扇出适用场景：

大数据处理的并行计算
负载均衡分发任务
多源数据聚合（如监控数据汇总）

Pipeline适用场景：

数据处理流水线（ETL过程）
图像处理的多阶段操作
编译器的工作流程

3. 总结

生产者-消费者：解耦生产消费速度，通过channel缓冲协调
扇出模式：一产多消，提高处理吞吐量，注意负载均衡
扇入模式：多产一消，聚合多源数据，注意channel管理
Pipeline模式：任务分解为阶段，每个阶段专注单一职责
错误处理：在并发场景中妥善处理错误，避免goroutine泄露

练习

并发模型（CSP理论、Goroutine轻量级线程）

1. 语法讲解

一、 代码逻辑分析

二、 为什么打印结果每次不一样，且执行顺序乱？

1. 主协程与子协程的抢占式调度

2. semaphore 的阻塞机制

3. fetchData 内部的耗时不确定

2. 应用场景

3. 总结

Channel（管道）

1. 语法讲解

2. 应用场景

1. go 关键字的作用：发射后不管

2. 这两个协程是“同时”运行的

3. 核心机制：Channel 的阻塞与唤醒

4. 一个至关重要的前提条件（必须注意）

总结

3. 总结

同步原语（sync.Mutex互斥锁、sync.WaitGroup等待组、sync.RWMutex读写锁）

1. 语法讲解

2. 应用场景

3. 总结

并发模式（生产者-消费者、扇入/扇出、Pipeline）

1. 语法讲解

2. 应用场景

3. 总结

一、代码逻辑分析

二、为什么打印结果每次不一样，且执行顺序乱？

2. `semaphore` 的阻塞机制

3. `fetchData` 内部的耗时不确定

1. `go` 关键字的作用：发射后不管