以通信来共享内存
扩展-一些名词了解
Linux、IPC（进程间通信）、进程（几个状态）、线程、同步、异步、信号、管道、socket、消息队列、字节流、结构化消息、通信、信号量、共享内存、内核空间、用户空间、PID、PPID、fork、COW 将进程细化为多个状态后，系统CPU就可以更加灵活地调度，进程下还有更细的线程，有N多状态， GO调度器实现一套机制，统一调度与分配这些CPU、进程、线程等资源，分为M(内核)、P(go上下文)、G（goroutine）三层 以上概念不学linux原理的话，听个名词就好，GO中重点学习两个点 1. 第三层G就是我们经常使用的goroutine协程，掌握goroutine的使用 2. GO支持的IPC方法有 信号、管道、socket，掌握这三种的实现方式
channel 定义
for 与 channel
go channel特性
同一时刻，仅有一个goroutine可以向该通道发送元素值，同时也仅有一个goroutine可以从该通道接收元素值，即通道是串行的。 通道中的元素值，严格按发送到该通道的先后顺序排列，最先发送到的元素值，一定最先被接收。等效于先进先出的消息队列。 通道中的元素值，具有原子性，不可被分割；每个元素值只能被一个goroutine接收，被接收后，立刻从通道中清除。 goroutine的执行与主程序是并行的，主程序结束时，还没有执行完毕的goroutine会被强制中止。 通道的传值是复制，不是引用。
通道初始化
通道未初始化时其值为nil，可以从nil中尝试接收元素，但会被永远阻塞 make(chan int) 初始化一个可以接收、发送int值类型的通道，无缓冲 make(chan int, 10) 初始化一个可以接收、发送int值类型的通道，可缓冲10个int值。 第11个int值向通道中发送时会被阻塞。 被缓冲的元素值，会严格按发送的顺序接收。
从通道中接收元素
关闭通道
长度与容量
单向通道
channel简单示例
channel同步
让后台协程有序执行
channel数组
package main
import (
"fmt"
"sync"
)
type worker struct {
in chan int
done func()
}
//“接收”方程序
func (wk *worker) doWork(id int){
//接收的确是按数组顺序顺序打印出来的，但这只是程序第一次运行的情况
//接收是在发送之前就以并行的方式运行起来了，之后数据中每个channel都一直处于阻塞等待状态
//也就是说数组中的每个channel谁先打印出数据，就表示该谁先发送数据（忽略channel传送数据时长差异）
for n := range wk.in {
fmt.Printf("第 %d 次接收的信息为 %c\n",id,n)
//通知主程序工作处理完毕
wk.done()
}
}
func createWorker(id int, wtg *sync.WaitGroup) worker{
//channel作为struct的一个属性
wk := worker{
//chan<-表示channel只用于发送数据，即输入
in : make(chan int),
done: func(){
wtg.Done()
},
}
//channel创建之后，就开始以并行的方式建立“接收”方
go wk.doWork(id)
return wk
}
func channel2(){
//WaitGroup的wait(在主程序中调用)与done(在与主程序并行执行的“接收”方中调用)的交互，
//可以达到等待所有channel运行完毕，再让主程序运行的效果
//而不是程序员猜想channel“接收”需要多少时间运行，
//然后去主程序中设置time.Sleep让主程序等待
var wtg sync.WaitGroup
//channel数组
var workers [8]worker
for i := 0; i < 8; i++{
//使用引用的方式传送参数，所有的channel公用一个WaitGroup
workers[i] = createWorker(i,&wtg)
}
//要一次性添加完要等待执行的channel个数
wtg.Add(16)
for i,worker := range workers {
worker.in <- 'a' + i
//wtg.Add(1) //这种方式会报错
}
for i,worker := range workers {
worker.in <- 'A' + i
}
//等待所有channel执行完毕，否则一直阻塞
wtg.Wait()
fmt.Println("所有channel执行完毕")
}
func main(){
channel2()
fmt.Println("主程序结束")
}
第 0 次接收的信息为 a
第 1 次接收的信息为 b
第 2 次接收的信息为 c
第 2 次接收的信息为 C
第 3 次接收的信息为 d
第 3 次接收的信息为 D
第 4 次接收的信息为 e
第 4 次接收的信息为 E
第 0 次接收的信息为 A
第 1 次接收的信息为 B
第 6 次接收的信息为 g
第 5 次接收的信息为 f
第 5 次接收的信息为 F
第 6 次接收的信息为 G
第 7 次接收的信息为 h
第 7 次接收的信息为 H
所有channel执行完毕
主程序结束
|
**结论：是的。** 无缓冲通道（`make(chan int)`）的发送与接收是**同步的**（Handshake / Rendezvous，握手）。
发送方必须等到有接收方准备好，才能成功将数据"交接"出去；否则发送方会被阻塞。
具体分两种情况：
1. **发送方先执行**：发送数据时若没有另一个 Goroutine 在等待接收，发送方会**被阻塞**，直到有接收方准备好。
2. **接收方先执行**：接收数据时若没有另一个 Goroutine 在等待发送，接收方会**被阻塞**，直到有发送方准备好。
**两者必须同时就绪，才能完成一次"握手"**，数据才会从发送方直接拷贝给接收方（不经过中间缓冲区）。
|
|
下面这段代码会引发**死锁（fatal error）**，因为发送方没有接收方：
```go
func main() {
ch := make(chan int)
ch <- 1 // 这里会永远阻塞，main goroutine 只有发送，没有接收
}
```
|
|
必须让接收方在另一个 Goroutine 中准备好（或由另一个 Goroutine 发送，主 Goroutine 接收）：
```go
func main() {
ch := make(chan int)
// 接收方先准备好（在另一个协程中）
go func() {
<- ch // 阻塞等待接收
fmt.Println("received")
}()
ch <- 1 // 此时发送方发现已有接收方在等待，立即发送成功
time.Sleep(time.Second) // 等待协程打印
}
```
|
|
**对比有缓冲通道** `make(chan int, 10)`：
- 缓冲区未满时，发送不需要接收方准备好
- 缓冲区非空时，接收不需要发送方准备好
- 只有缓冲区满 / 空时才会阻塞
**总结**：无缓冲通道强调"同步"，**发送成功的前提是必须有接收方正在等待**（反之亦然）。
因此无缓冲通道天然适合用作 Goroutine 之间的**信号传递**或**同步屏障**。
|
|
以 `gowks/pkg/drs/c2p.go` 第 48 行（DescribeTable 函数首参）为例：
```go
func DescribeTable(ctx context.Context, db *sql.DB, table string) ([]ColumnDef, error) {
rows, err := db.QueryContext(ctx, "DESCRIBE "+qt)
...
}
```
`context.Context` 是 Go 标准库接口，用于**跨 API 边界、跨 goroutine** 传递三样东西：
1. **取消信号**（cancel）：外部一声令下，整条调用链同时收工
2. **截止时间 / 超时**（deadline / timeout）：到点自动取消
3. **请求级元数据**（value）：如 trace id（谨慎使用）
接口定义只有四个方法：
```go
type Context interface {
Deadline() (deadline time.Time, ok bool) // 截止时间
Done() <-chan struct{} // 取消/超时时被关闭的通道（只读）
Err() error // 取消原因：Canceled 或 DeadlineExceeded
Value(key any) any // 请求级元数据
}
```
**创建与派生**（ctx 是一棵树，根在上、枝叶向下）：
| 函数 | 作用 |
|------|------|
| `context.Background()` | 根 ctx，永不取消（程序入口用） |
| `context.TODO()` | 占位根 ctx（还不知道该传什么时） |
| `context.WithCancel(parent)` | 派生可手动取消的 ctx，返回 `cancel()` |
| `context.WithTimeout(parent, d)` | 派生超时自动取消的 ctx |
| `context.WithDeadline(parent, t)` | 派生到点自动取消的 ctx |
| `context.WithValue(parent, k, v)` | 派生携带元数据的 ctx |
**约定俗成**：ctx 作为函数**第一个参数**，命名 `ctx`；显式逐层传递，不存全局变量。
|
|
c2p.go 中 ctx 贯穿整条调用链，共四处典型用法：
**1. 透传给 database/sql（c2p.go:53）**
```go
rows, err := db.QueryContext(ctx, "DESCRIBE "+qt)
```
`QueryContext` 把 ctx 交给数据库驱动：ctx 一旦取消，**在途 SQL 立即中断、连接被释放**；若用不带 ctx 的 `db.Query`，卡住的查询无法中止。
**2. 派生超时 ctx（c2p.go:671，schemaFor 缓存表结构）**
```go
qctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
defs, err := DescribeTable(qctx, l.db, table)
```
父 ctx 不取消、子查询最多等 30 秒——**超时只收窄不放宽**。
**3. 顶层入口一路透传（转换阶段）**
```
ConvertCSVToParquet(ctx, ...) // 入口
→ convertSources(ctx, ...) // worker 池
→ convertOneSource(ctx, ...) // 单文件转换
```
**4. 内存直通导入（c2p.go:693/796）**
```go
func (l *C2PMemLoader) Run(ctx context.Context, task Task, ...) (res Result) {
...
lr, err := l.imp.streamLoadPayload(ctx, streamJob{...}, progress)
}
```
HTTP 的 Stream Load 请求也挂在**同一棵取消树**上——Ctrl+C 一个信号，SQL、HTTP、文件转换全部收工。
|
|
ctx 是**广播式取消**：一个 `cancel()`，所有从它派生的 ctx 的 `Done()` 通道**同时关闭**——天然适合 worker 池"一键停止"。c2p.go 的 convertSources（14 协程并发转换 315GB CSV）就是标准范本，三处配合：
**1. 派发协程：select 双路，取消即停止派活（c2p.go:453）**
```go
go func() {
defer close(jobs)
for i, sec := range secs {
select {
case jobs <- job{sec: sec, tag: tags[i]}: // 派活
case <-ctx.Done(): // 取消：立即停派
return
}
}
}()
```
**2. worker 接活前先看取消（c2p.go:428）**
```go
for j := range jobs {
if ctx.Err() != nil { // 已被取消：不再接新活
return
}
chunks, rows, bytes, err := convertOneSource(ctx, j.sec, j.tag, defs, opt, wopt, sep)
...
}
```
**3. 单文件长循环每块边界检查（c2p.go:506）**
```go
for {
if ctx.Err() != nil {
return nil, 0, 0, ctx.Err() // 300 万行一块的边界上尽早退出
}
...
}
```
**要点总结**：
- 取消后，派发方不再喂活、worker 不再接活、在途的长任务在**块边界**尽快退出——三层都拦，取消延迟最小
- `ctx.Err()` 轮询（非阻塞）与 `select <-ctx.Done()`（等待）是两种消费姿势，按场景选
- 最后 `wg.Wait()` 等所有 worker 收尾，再统一上报 `ctx.Err()`
**本质**：`Done()` 返回的就是一个 `<-chan struct{}`，取消即 `close`——channel 一关，所有接收方同时被唤醒。这正是"以通信来共享内存"，与前面 channel 同步一节完全呼应：**context 是建立在 channel 之上的广播机制**。
|
|
**1. 派生必须 `defer cancel()`**
```go
qctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel() // 即使提前返回也要 cancel，否则父 ctx 资源泄漏到超时为止
```
`WithTimeout` / `WithCancel` 返回的 `cancel` 不调，子 ctx 会一直挂在父树上，直到超时到期才释放。
**2. 非阻塞检查 vs 阻塞等待**
```go
if ctx.Err() != nil { return ctx.Err() } // 轮询：长循环边界用
select { // 等待：和 channel 收发竞争时用
case jobs <- j:
case <-ctx.Done():
return
}
```
**3. 根 ctx 只在程序入口创建**
`context.Background()` / `TODO()` 只出现在 main 或请求入口，之后**逐层透传**，底层函数不自己造根（造了就切断了取消链）。
**4. 不要把 ctx 存进 struct 长期持有**
c2p.go 的 `C2PMemLoader` struct 里只有 cfg/db/emit，**没有存 ctx**——ctx 跟随每次 `Run(ctx, task)` 调用传入，一个任务一条生命周期，互不串扰。
**5. `ctx.Value` 不是传参工具**
多并发场景中往 Value 里塞业务数据是反模式——类型不安全、无编译期检查。**参数能显式传的，就别塞 ctx**；Value 只放真正的请求级元数据（trace id 等）。
**6. 取消是协作式的，不是强杀**
`cancel()` 不会杀掉 goroutine，只是把 Done() 通道关闭。被调用方必须**主动检查**（Err 轮询或 select Done）并自愿退出——c2p.go 三层检查就是这个协作关系的落地。
|