Go学习笔记-Channels

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

package main

import (
    "fmt"
    "time"
)

func main() {
    // 1. 创建channel
    fmt.Println("=== 基本Channel操作 ===")
    
    // 无缓冲channel
    ch1 := make(chan int)
    
    // 带缓冲channel
    ch2 := make(chan string, 3)
    
    // 只读channel
    var readOnly <-chan int = ch1
    
    // 只写channel
    var writeOnly chan<- int = ch1
    
    fmt.Printf("无缓冲channel: %T\n", ch1)
    fmt.Printf("带缓冲channel: %T\n", ch2)
    fmt.Printf("只读channel: %T\n", readOnly)
    fmt.Printf("只写channel: %T\n", writeOnly)
    
    // 2. 基本发送和接收
    fmt.Println("\n=== 基本发送和接收 ===")
    
    // 启动goroutine发送数据
    go func() {
        ch1 <- 42
        fmt.Println("发送了数据: 42")
    }()
    
    // 接收数据
    value := <-ch1
    fmt.Printf("接收到数据: %d\n", value)
    
    // 3. 带缓冲channel的使用
    fmt.Println("\n=== 带缓冲Channel ===")
    
    // 可以发送多个值而不阻塞
    ch2 <- "hello"
    ch2 <- "world"
    ch2 <- "go"
    
    fmt.Printf("缓冲区长度: %d, 容量: %d\n", len(ch2), cap(ch2))
    
    // 接收数据
    for i := 0; i < 3; i++ {
        msg := <-ch2
        fmt.Printf("接收到: %s\n", msg)
    }
    
    // 4. 关闭channel
    fmt.Println("\n=== 关闭Channel ===")
    
    ch3 := make(chan int, 2)
    
    // 发送数据
    ch3 <- 1
    ch3 <- 2
    
    // 关闭channel
    close(ch3)
    
    // 从关闭的channel接收数据
    for {
        value, ok := <-ch3
        if !ok {
            fmt.Println("Channel已关闭")
            break
        }
        fmt.Printf("接收到: %d\n", value)
    }
    
    // 5. 使用range遍历channel
    fmt.Println("\n=== Range遍历Channel ===")
    
    ch4 := make(chan int, 5)
    
    // 发送数据并关闭
    go func() {
        for i := 1; i <= 5; i++ {
            ch4 <- i
        }
        close(ch4)
    }()
    
    // 使用range遍历
    for value := range ch4 {
        fmt.Printf("Range接收到: %d\n", value)
    }
    
    // 6. 非阻塞操作
    fmt.Println("\n=== 非阻塞操作 ===")
    
    ch5 := make(chan int, 1)
    
    // 非阻塞发送
    select {
    case ch5 <- 100:
        fmt.Println("成功发送数据")
    default:
        fmt.Println("发送失败，channel满了")
    }
    
    // 非阻塞接收
    select {
    case value := <-ch5:
        fmt.Printf("成功接收数据: %d\n", value)
    default:
        fmt.Println("接收失败，channel空了")
    }
    
    // 再次尝试非阻塞接收
    select {
    case value := <-ch5:
        fmt.Printf("成功接收数据: %d\n", value)
    default:
        fmt.Println("接收失败，channel空了")
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327

package main

import (
    "fmt"
    "math/rand"
    "time"
)

// 1. 模拟网络请求
func fetchData(url string, ch chan<- string) {
    // 模拟不同的响应时间
    delay := time.Duration(rand.Intn(1000)) * time.Millisecond
    time.Sleep(delay)
    
    ch <- fmt.Sprintf("数据来自 %s (延迟: %v)", url, delay)
}

// 2. 心跳检测
func heartbeat(interval time.Duration) <-chan string {
    ch := make(chan string)
    
    go func() {
        defer close(ch)
        ticker := time.NewTicker(interval)
        defer ticker.Stop()
        
        for i := 1; i <= 5; i++ {
            select {
            case <-ticker.C:
                ch <- fmt.Sprintf("心跳 %d", i)
            }
        }
    }()
    
    return ch
}

// 3. 超时处理
func operationWithTimeout(duration time.Duration) <-chan string {
    result := make(chan string, 1)
    
    go func() {
        // 模拟长时间操作
        time.Sleep(2 * time.Second)
        result <- "操作完成"
    }()
    
    return result
}

// 4. 多路复用
func multiplexer(inputs ...<-chan string) <-chan string {
    output := make(chan string)
    
    go func() {
        defer close(output)
        
        for {
            select {
            case msg, ok := <-inputs[0]:
                if !ok {
                    inputs[0] = nil
                } else {
                    output <- fmt.Sprintf("输入1: %s", msg)
                }
            case msg, ok := <-inputs[1]:
                if !ok {
                    inputs[1] = nil
                } else {
                    output <- fmt.Sprintf("输入2: %s", msg)
                }
            case msg, ok := <-inputs[2]:
                if !ok {
                    inputs[2] = nil
                } else {
                    output <- fmt.Sprintf("输入3: %s", msg)
                }
            }
            
            // 检查是否所有输入都已关闭
            allClosed := true
            for _, ch := range inputs {
                if ch != nil {
                    allClosed = false
                    break
                }
            }
            if allClosed {
                break
            }
        }
    }()
    
    return output
}

func main() {
    // 1. 基本select使用
    fmt.Println("=== 基本Select使用 ===")
    
    ch1 := make(chan string)
    ch2 := make(chan string)
    
    // 启动两个goroutine
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "来自channel1的消息"
    }()
    
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "来自channel2的消息"
    }()
    
    // 使用select等待第一个可用的channel
    select {
    case msg1 := <-ch1:
        fmt.Printf("收到: %s\n", msg1)
    case msg2 := <-ch2:
        fmt.Printf("收到: %s\n", msg2)
    case <-time.After(3 * time.Second):
        fmt.Println("超时了")
    }
    
    // 2. 竞争条件 - 最快响应
    fmt.Println("\n=== 竞争条件 ===")
    
    urls := []string{
        "https://api1.example.com",
        "https://api2.example.com",
        "https://api3.example.com",
    }
    
    results := make([]chan string, len(urls))
    for i, url := range urls {
        results[i] = make(chan string, 1)
        go fetchData(url, results[i])
    }
    
    // 等待最快的响应
    select {
    case result := <-results[0]:
        fmt.Printf("最快响应: %s\n", result)
    case result := <-results[1]:
        fmt.Printf("最快响应: %s\n", result)
    case result := <-results[2]:
        fmt.Printf("最快响应: %s\n", result)
    case <-time.After(2 * time.Second):
        fmt.Println("所有请求都超时了")
    }
    
    // 3. 超时模式
    fmt.Println("\n=== 超时模式 ===")
    
    operation := operationWithTimeout(1 * time.Second)
    
    select {
    case result := <-operation:
        fmt.Printf("操作结果: %s\n", result)
    case <-time.After(1 * time.Second):
        fmt.Println("操作超时")
    }
    
    // 4. 心跳检测
    fmt.Println("\n=== 心跳检测 ===")
    
    heartbeatCh := heartbeat(500 * time.Millisecond)
    
    for {
        select {
        case beat, ok := <-heartbeatCh:
            if !ok {
                fmt.Println("心跳检测结束")
                goto nextSection
            }
            fmt.Printf("收到: %s\n", beat)
        case <-time.After(1 * time.Second):
            fmt.Println("心跳超时")
            goto nextSection
        }
    }
    
nextSection:
    // 5. 多路复用
    fmt.Println("\n=== 多路复用 ===")
    
    input1 := make(chan string)
    input2 := make(chan string)
    input3 := make(chan string)
    
    // 启动数据生产者
    go func() {
        defer close(input1)
        for i := 1; i <= 3; i++ {
            input1 <- fmt.Sprintf("A%d", i)
            time.Sleep(300 * time.Millisecond)
        }
    }()
    
    go func() {
        defer close(input2)
        for i := 1; i <= 3; i++ {
            input2 <- fmt.Sprintf("B%d", i)
            time.Sleep(500 * time.Millisecond)
        }
    }()
    
    go func() {
        defer close(input3)
        for i := 1; i <= 3; i++ {
            input3 <- fmt.Sprintf("C%d", i)
            time.Sleep(700 * time.Millisecond)
        }
    }()
    
    // 多路复用
    output := multiplexer(input1, input2, input3)
    
    for msg := range output {
        fmt.Printf("多路复用输出: %s\n", msg)
    }
    
    // 6. 优雅关闭
    fmt.Println("\n=== 优雅关闭 ===")
    
    done := make(chan bool)
    data := make(chan int)
    
    // 数据生产者
    go func() {
        defer close(data)
        for i := 1; i <= 10; i++ {
            select {
            case data <- i:
                fmt.Printf("发送数据: %d\n", i)
                time.Sleep(200 * time.Millisecond)
            case <-done:
                fmt.Println("生产者收到关闭信号")
                return
            }
        }
    }()
    
    // 数据消费者
    go func() {
        for {
            select {
            case value, ok := <-data:
                if !ok {
                    fmt.Println("数据channel已关闭")
                    done <- true
                    return
                }
                fmt.Printf("接收数据: %d\n", value)
                
                // 模拟在第5个数据时请求关闭
                if value == 5 {
                    fmt.Println("请求关闭...")
                    done <- true
                    return
                }
            }
        }
    }()
    
    // 等待完成
    <-done
    fmt.Println("程序优雅关闭")
    
    // 7. 扇入模式
    fmt.Println("\n=== 扇入模式 ===")
    
    fanIn := func(input1, input2 <-chan string) <-chan string {
        output := make(chan string)
        
        go func() {
            defer close(output)
            for {
                select {
                case msg, ok := <-input1:
                    if !ok {
                        input1 = nil
                    } else {
                        output <- msg
                    }
                case msg, ok := <-input2:
                    if !ok {
                        input2 = nil
                    } else {
                        output <- msg
                    }
                }
                
                if input1 == nil && input2 == nil {
                    break
                }
            }
        }()
        
        return output
    }
    
    ch1 = make(chan string)
    ch2 = make(chan string)
    
    go func() {
        defer close(ch1)
        for i := 1; i <= 3; i++ {
            ch1 <- fmt.Sprintf("输入1-%d", i)
            time.Sleep(300 * time.Millisecond)
        }
    }()
    
    go func() {
        defer close(ch2)
        for i := 1; i <= 3; i++ {
            ch2 <- fmt.Sprintf("输入2-%d", i)
            time.Sleep(400 * time.Millisecond)
        }
    }()
    
    merged := fanIn(ch1, ch2)
    
    for msg := range merged {
        fmt.Printf("扇入输出: %s\n", msg)
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449

package main

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

// 1. 工作池模式
type Job struct {
    ID   int
    Data string
}

type Result struct {
    Job    Job
    Output string
    Error  error
}

type WorkerPool struct {
    jobs    chan Job
    results chan Result
    workers int
    wg      sync.WaitGroup
}

func NewWorkerPool(workers int, jobBuffer int) *WorkerPool {
    return &WorkerPool{
        jobs:    make(chan Job, jobBuffer),
        results: make(chan Result, jobBuffer),
        workers: workers,
    }
}

func (wp *WorkerPool) Start() {
    for i := 0; i < wp.workers; i++ {
        wp.wg.Add(1)
        go wp.worker(i)
    }
}

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    
    for job := range wp.jobs {
        fmt.Printf("Worker %d 处理任务 %d\n", id, job.ID)
        
        // 模拟工作
        time.Sleep(time.Duration(job.ID*100) * time.Millisecond)
        
        result := Result{
            Job:    job,
            Output: fmt.Sprintf("处理结果: %s", job.Data),
            Error:  nil,
        }
        
        wp.results <- result
    }
}

func (wp *WorkerPool) Submit(job Job) {
    wp.jobs <- job
}

func (wp *WorkerPool) Close() {
    close(wp.jobs)
    wp.wg.Wait()
    close(wp.results)
}

func (wp *WorkerPool) Results() <-chan Result {
    return wp.results
}

// 2. 请求-响应模式
type Request struct {
    ID       int
    Data     string
    Response chan Response
}

type Response struct {
    ID     int
    Result string
    Error  error
}

func requestResponseServer(requests <-chan Request) {
    for req := range requests {
        go func(r Request) {
            // 处理请求
            time.Sleep(time.Duration(r.ID*50) * time.Millisecond)
            
            response := Response{
                ID:     r.ID,
                Result: fmt.Sprintf("处理了请求: %s", r.Data),
                Error:  nil,
            }
            
            r.Response <- response
            close(r.Response)
        }(req)
    }
}

// 3. 发布-订阅模式
type PubSub struct {
    subscribers map[string][]chan string
    mutex       sync.RWMutex
}

func NewPubSub() *PubSub {
    return &PubSub{
        subscribers: make(map[string][]chan string),
    }
}

func (ps *PubSub) Subscribe(topic string) <-chan string {
    ps.mutex.Lock()
    defer ps.mutex.Unlock()
    
    ch := make(chan string, 10)
    ps.subscribers[topic] = append(ps.subscribers[topic], ch)
    return ch
}

func (ps *PubSub) Publish(topic, message string) {
    ps.mutex.RLock()
    defer ps.mutex.RUnlock()
    
    for _, ch := range ps.subscribers[topic] {
        select {
        case ch <- message:
        default:
            // 如果channel满了，跳过这个订阅者
        }
    }
}

func (ps *PubSub) Unsubscribe(topic string, ch <-chan string) {
    ps.mutex.Lock()
    defer ps.mutex.Unlock()
    
    subscribers := ps.subscribers[topic]
    for i, subscriber := range subscribers {
        if subscriber == ch {
            // 移除订阅者
            ps.subscribers[topic] = append(subscribers[:i], subscribers[i+1:]...)
            close(subscriber)
            break
        }
    }
}

// 4. 限流器
type RateLimiter struct {
    tokens chan struct{}
    ticker *time.Ticker
}

func NewRateLimiter(rate int, burst int) *RateLimiter {
    rl := &RateLimiter{
        tokens: make(chan struct{}, burst),
        ticker: time.NewTicker(time.Second / time.Duration(rate)),
    }
    
    // 初始填满令牌桶
    for i := 0; i < burst; i++ {
        rl.tokens <- struct{}{}
    }
    
    // 定期添加令牌
    go func() {
        for range rl.ticker.C {
            select {
            case rl.tokens <- struct{}{}:
            default:
                // 令牌桶满了
            }
        }
    }()
    
    return rl
}

func (rl *RateLimiter) Allow() bool {
    select {
    case <-rl.tokens:
        return true
    default:
        return false
    }
}

func (rl *RateLimiter) Wait(ctx context.Context) error {
    select {
    case <-rl.tokens:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (rl *RateLimiter) Close() {
    rl.ticker.Stop()
    close(rl.tokens)
}

// 5. 断路器模式
type CircuitBreaker struct {
    maxFailures int
    resetTime   time.Duration
    failures    int
    lastFailure time.Time
    state       string // "closed", "open", "half-open"
    mutex       sync.Mutex
}

func NewCircuitBreaker(maxFailures int, resetTime time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        maxFailures: maxFailures,
        resetTime:   resetTime,
        state:       "closed",
    }
}

func (cb *CircuitBreaker) Call(fn func() error) error {
    cb.mutex.Lock()
    defer cb.mutex.Unlock()
    
    // 检查是否可以从open状态转换到half-open状态
    if cb.state == "open" && time.Since(cb.lastFailure) > cb.resetTime {
        cb.state = "half-open"
        cb.failures = 0
    }
    
    // 如果断路器是开启状态，直接返回错误
    if cb.state == "open" {
        return fmt.Errorf("断路器开启，拒绝请求")
    }
    
    // 执行函数
    err := fn()
    
    if err != nil {
        cb.failures++
        cb.lastFailure = time.Now()
        
        if cb.failures >= cb.maxFailures {
            cb.state = "open"
        }
        
        return err
    }
    
    // 成功执行，重置失败计数
    if cb.state == "half-open" {
        cb.state = "closed"
    }
    cb.failures = 0
    
    return nil
}

func main() {
    // 1. 工作池模式
    fmt.Println("=== 工作池模式 ===")
    
    pool := NewWorkerPool(3, 10)
    pool.Start()
    
    // 提交任务
    go func() {
        for i := 1; i <= 10; i++ {
            job := Job{
                ID:   i,
                Data: fmt.Sprintf("任务数据-%d", i),
            }
            pool.Submit(job)
        }
        pool.Close()
    }()
    
    // 收集结果
    for result := range pool.Results() {
        fmt.Printf("任务 %d 完成: %s\n", result.Job.ID, result.Output)
    }
    
    // 2. 请求-响应模式
    fmt.Println("\n=== 请求-响应模式 ===")
    
    requests := make(chan Request, 10)
    go requestResponseServer(requests)
    
    // 发送请求
    for i := 1; i <= 5; i++ {
        responseCh := make(chan Response, 1)
        req := Request{
            ID:       i,
            Data:     fmt.Sprintf("请求数据-%d", i),
            Response: responseCh,
        }
        
        requests <- req
        
        // 等待响应
        response := <-responseCh
        fmt.Printf("收到响应 %d: %s\n", response.ID, response.Result)
    }
    
    close(requests)
    
    // 3. 发布-订阅模式
    fmt.Println("\n=== 发布-订阅模式 ===")
    
    pubsub := NewPubSub()
    
    // 创建订阅者
    sub1 := pubsub.Subscribe("news")
    sub2 := pubsub.Subscribe("news")
    sub3 := pubsub.Subscribe("sports")
    
    // 启动订阅者goroutine
    go func() {
        for msg := range sub1 {
            fmt.Printf("订阅者1收到新闻: %s\n", msg)
        }
    }()
    
    go func() {
        for msg := range sub2 {
            fmt.Printf("订阅者2收到新闻: %s\n", msg)
        }
    }()
    
    go func() {
        for msg := range sub3 {
            fmt.Printf("订阅者3收到体育: %s\n", msg)
        }
    }()
    
    // 发布消息
    pubsub.Publish("news", "重要新闻1")
    pubsub.Publish("news", "重要新闻2")
    pubsub.Publish("sports", "体育新闻1")
    
    time.Sleep(100 * time.Millisecond)
    
    // 4. 限流器
    fmt.Println("\n=== 限流器 ===")
    
    limiter := NewRateLimiter(2, 5) // 每秒2个请求，突发5个
    defer limiter.Close()
    
    // 测试限流
    for i := 1; i <= 10; i++ {
        if limiter.Allow() {
            fmt.Printf("请求 %d 通过\n", i)
        } else {
            fmt.Printf("请求 %d 被限流\n", i)
        }
        time.Sleep(200 * time.Millisecond)
    }
    
    // 5. 断路器模式
    fmt.Println("\n=== 断路器模式 ===")
    
    cb := NewCircuitBreaker(3, 2*time.Second)
    
    // 模拟可能失败的操作
    failingOperation := func() error {
        // 前几次调用会失败
        if time.Now().Unix()%2 == 0 {
            return fmt.Errorf("操作失败")
        }
        return nil
    }
    
    // 测试断路器
    for i := 1; i <= 10; i++ {
        err := cb.Call(failingOperation)
        if err != nil {
            fmt.Printf("调用 %d 失败: %v\n", i, err)
        } else {
            fmt.Printf("调用 %d 成功\n", i)
        }
        time.Sleep(300 * time.Millisecond)
    }
    
    // 6. 管道取消
    fmt.Println("\n=== 管道取消 ===")
    
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()
    
    // 创建可取消的管道
    numbers := make(chan int)
    squares := make(chan int)
    
    // 数字生成器
    go func() {
        defer close(numbers)
        for i := 1; i <= 10; i++ {
            select {
            case numbers <- i:
                fmt.Printf("生成数字: %d\n", i)
                time.Sleep(300 * time.Millisecond)
            case <-ctx.Done():
                fmt.Println("数字生成器被取消")
                return
            }
        }
    }()
    
    // 平方计算器
    go func() {
        defer close(squares)
        for {
            select {
            case num, ok := <-numbers:
                if !ok {
                    return
                }
                squares <- num * num
                fmt.Printf("计算平方: %d -> %d\n", num, num*num)
            case <-ctx.Done():
                fmt.Println("平方计算器被取消")
                return
            }
        }
    }()
    
    // 结果消费者
    for {
        select {
        case square, ok := <-squares:
            if !ok {
                fmt.Println("管道处理完成")
                return
            }
            fmt.Printf("最终结果: %d\n", square)
        case <-ctx.Done():
            fmt.Println("管道被超时取消")
            return
        }
    }
}