通道

发表于 2026/01/20 更新于 2026/09/12

作者 lfj

36 分钟阅读

内存泄露

对于一个channel来说，如果没有任何goroutine(轻量级线程)引用，gc会对其进行回收，不会引起内存泄露。

  
// goroutine
go func() {
    fmt.Println("hello from a goroutine")
}()

而当goroutine处于接收或发送阻塞状态，channel处于空或满状态时，一直得不到改变，gc则无法回收这类一直处于等待队列中的goroutine，引起内存泄露。

引起内存泄露的几种情况：

channel缓存队列已满，接收者goroutine退出，发送者阻塞且所在goroutine无法退出
channel缓存队列未满，发送者goroutine退出，接收者阻塞且所在goroutine无法退出
channel未初始化

  
//情形1：  channel缓存队列已满，接收者goroutine退出，发送者阻塞且所在goroutine无法退出
func TestName(t *testing.T) {
	channel := make(chan int, 5)
	//sender
	go func() {
		for i := 0; i < 10; i++ {
			channel <- i //阻塞
			time.Sleep(10 * time.Millisecond)
			fmt.Println("sender count:", i)
		}
		fmt.Println("sender exit.") //sender 无法退出
	}()

	//receiver
	go func() {
		i := <-channel //receiver 接收一个退出
		fmt.Println("receive:", i)
		fmt.Println("receiver exit.")
	}()

	for {
		time.Sleep(10 * time.Millisecond)
	}
}

/**
=== RUN   TestName
receive: 0
receiver exit.
sender count: 0
sender count: 1
sender count: 2
sender count: 3
sender count: 4
sender count: 5


*/
// 解决方式
func TestName1(t *testing.T) {
    channel := make(chan int, 5)

    // sender
    go func() {
        channel <- 1
        close(channel)          // 关键：关闭 channel
        fmt.Println("sender exit.")
    }()

    // receiver
    go func() {
        for v := range channel { // range 会在 channel 关闭且读完数据后自动退出
            fmt.Println("receive:", v)
        }
        fmt.Println("receiver exit.")
    }()

    time.Sleep(100 * time.Millisecond) // 等它们跑完
}



// 情形2：  channel缓存队列未满，发送者goroutine退出，接收者阻塞且所在goroutine无法退出
func TestName1(t *testing.T) {
	channel := make(chan int, 5)
	//sender
	go func() {
		channel <- 1 //发一个退出
		fmt.Println("sender exit.")
	}()

	//receiver
	go func() {
		for i := 0; i < 10; i++ {
			i := <-channel //阻塞
			fmt.Println("receive:", i)
		}
		fmt.Println("receiver exit.")
	}()

	for {
		time.Sleep(10 * time.Millisecond)
	}

}

/**
=== RUN   TestName1
sender exit.
receive: 1

*/

// 解决：取消信号. 加一个通知另一段退出的stopChan
func TestName2(t *testing.T) {
	channel := make(chan int, 5)
	stopChan := make(chan struct{}) //无缓冲的channel作为信号量

	//sender
	go func() {
		for i := 0; i < 10; i++ {
			select {
			case <-stopChan: //收到close信号，退出。多个sender时，每个goroutine也都能收到
				fmt.Println("sender exit.")
				return
			case channel <- i:
				time.Sleep(10 * time.Millisecond)
				fmt.Println("sender count:", i)
			}
		}
		fmt.Println("sender exit.")
	}()

	//receiver
	go func() {
		i := <-channel
		close(stopChan)
		fmt.Println("receive:", i)
		fmt.Println("receiver exit.")
	}()

	for {
		time.Sleep(10 * time.Millisecond)
	}

}

备注：

从一个关闭的channel仍能读出数据，当缓冲队列还有数据时，会返回正常值；缓冲队列为空时，则返回channel已关闭。
向一个关闭的channel发送数据则会报Panic

阻塞判断

  
for skuMsg := range skuChan {  // <--- 这里会一直等待直到channel被关闭
    // 处理每个sku消息
}
//当 channel 被关闭且所有数据被消费后，循环自然结束

三种状态下的行为

数据可用: 立即获取数据并执行循环体
数据未到达: 阻塞等待直到有数据或 channel 关闭
channel 关闭且空: 循环正常退出
同步机制

生产者-消费者模型

  
生产者通过 ch <- data 发送数据
消费者通过 for range 接收数据
自动实现了生产者和消费者的同步

关闭信号传递

close(channel) 既是关闭信号也是完成信号
每个 channel 都有 defer close(ch) 保证最终会被关闭range, 循环会在 channel 关闭时自动结束,即使暂时没有数据，range 也会等待直到有数据或者 channel 被关闭
消费者通过检测 channel 状态知道生产已完成

内存可见性保障

Go 的 channel 实现提供了必要的内存屏障
确保数据在 goroutine 间正确传递
这种设计使得 for range 循环能够优雅地处理并发数据流，无需额外的同步原语即可实现完整的生产者-消费者模式。

带缓存和无缓存

无缓冲（N=0）——发送端必阻塞

  
ch := make(chan item) // 容量 0
ch <- it         // 写端会卡住，直到有人 `<-ch` 读走上一个

特点：

同步操作：发送和接收必须同时就绪才能完成操作
阻塞行为：发送操作会阻塞直到有接收者准备就绪
接收操作：也会阻塞直到有发送者准备就绪
保证顺序：发送和接收严格配对，确保数据传递的同步性

使用场景：

需要严格的同步通信生产者消费者模式中需要精确控制协程间协调操作

有缓冲（N>0）——发送端先丢进队列，立即返回

  
ch := make(chan item, 2) // 容量 2
ch <- it1                // 缓冲区空，直接拷贝进去，写端立即返回
ch <- it2                // 缓冲区还有 1 个空位，再拷贝，立即返回

特点：

异步操作：发送方可以在缓存未满时立即发送, 写不阻塞, 直到满了
非阻塞发送：只要缓存未满，发送不会阻塞
非阻塞接收：只要缓存非空，接收不会阻塞
缓冲作用：可以暂时存储数据，平滑生产者和消费者的速率差异

使用场景：

数据流处理中的临时存储
生产者和消费者速率不匹配的情况
避免不必要的阻塞，提高并发性能

案例

  
// 写入
func notifySaveOneIDBatchTasks(payloads []*saveOneIDBatchTask, result map[string]string, err error) {
        for _, p := range payloads {
            value := ""
            if result != nil {
                if p.direction == "oneid_to_openuid" {
                    value = result[p.req.OneID]
                } else {
                    value = result[p.req.BuyerOpenUID]
                }
            }
            //写入，不阻塞（缓冲=1)
            p.done <- oneIDBatchResult{value: value, err: err}
            close(p.done)
        }
}

// 读取
func waitOneIDBatchResult(ctx context.Context, done <-chan oneIDBatchResult) (string, error) {
        // 阻塞等待结果
        select {
        case <-ctx.Done():
            return "", ctx.Err()
        case result := <-done:
            return result.value, result.err
        }
}

  
无缓冲 channel 的 send 必须有 receiver 同时准备好才能完成（handshake）。

      done := make(chan oneIDBatchResult) // 无缓冲
      p.done <- result // 阻塞！等 receiver 来读
      close(p.done)    // send 完成后才能执行到这里

      如果 waitOneIDBatchResult 已经因为 ctx.Done() 退出了，没人读这个 channel，p.done <- result 就永远卡住——goroutine 泄漏。                                        

缓冲=1：

      done := make(chan oneIDBatchResult, 1) // 缓冲=1
      p.done <- result // 直接写入缓冲区，不等 receiver，立刻返回
      close(p.done)    // 立刻执行，关闭 channel

      即使 receiver 已经走了，send 也能成功（值进缓冲区, 默认会把缓存区只读完），close 也立刻完成。没有阻塞，没有泄漏。

无缓存, close的时候, 如果值进缓冲区, 默认会把缓存区只读完

// 取消后：先非阻塞地 drain 掉缓冲里的消息，再退出
case <-ctx.Done():
    for {
        select {
        case message := <-w.Channel:
            w.Handle(ctx, message)  // 尽量处理完堆积
        default:
            return nil  // 没有堆积了，退出
        }
    }

写法

入参:

  
onsaleCh chan <- IterGoodsResult

出参

chan SupplierSkuMsg

写入

  
ch <- SupplierSkuMsg{
    SkuId:     resWare.GoodsID,
    IsMainSku: true,
    Price:     resWare.Price,
}

读取

  
// 方法1
for skuMsg := range ch { 
    //… 
}

// 方法2
for {
    select {
    case <-ctx.Done():
        return ctx.Err()
    case job := <-jobCh:
         //… 
    }
}

// 方式3,  阻塞「等一个结果」或「等一个完成/就绪信号」 的同步场景
<-jobCh

搭配select

使用的典型场景可以一句话概括：“当你想同时等好几个 channel 的收发事件，谁先到就处理谁” 的时候，就必须用 select。

基础

  
// range
for skuMsg := range ch { //… }  

// 等价于
for {
    v, ok := <-ch
    if !ok {          // 只有 ch 被关闭才能退出
        break
    }
    // …
}

例子

  
{
    for skuMsg := range skuChan1 { … }   // ① 把 ch1 全部消费完
    for skuMsg := range skuChan2 { … }   // ② 再把 ch2 全部消费完
}

  
for {
    // 谁先到用谁
    select {
    case skuMsg, ok := <-skuChan1:
        if !ok { skuChan1 = nil; continue }
        ...
    case skuMsg, ok := <-skuChan2:
        if !ok { skuChan2 = nil; continue }
        ...
    }
}

主动取消（context.Done）

  
//只处理一条就结束 → 不用 for
select {  // select不带default是阻塞的
case <-ctx.Done():
    return ctx.Err()
case job := <-jobCh:
    process(job)   // 只处理一次就返回
}

//想循环不断地从 jobCh 里拿任务，直到上游取消 → 必须包 for
for {
    select {
    case <-ctx.Done():
        return ctx.Err()
    case job := <-jobCh:
        process(job)   // 每条都处理
    }
}

多路等待（multiplex）

  
// 读
select {
case v := <-ch1:
    fmt.Println("ch1 先到:", v)
case v := <-ch2:
    fmt.Println("ch2 先到:", v)
}

非阻塞收发（带 default）

消息必须立刻发出去，否则就直接丢，绝对不能阻塞上游的场景

**default 它只有在 channel 有缓冲（≥1）, **通道暂时为空（触发 default）

  
ch1 := make(chan string, 1)
ch2 := make(chan string, 1)

// 写, 如果select没套循环，它只执行一次
select {
    case ch1 <- msg: //投递
    case ch2 <- msg: //投递
    default: // 实现select非阻塞
      // 缓冲区两个都满，丢弃
}

// 注意
	for i := 0; i < p.batchSize; i++ {
    select {
    default:
       break  // break 只跳出 select，不会跳出for
    }
}

例子

 // 调用方（同步视角）
  func doSubmitSaveOneIDBatch(ctx context.Context, req *proto.SaveOneIDInfoReq, direction string) (string, error) {
      task := &saveOneIDBatchTask{
          ...
          done: make(chan oneIDBatchResult, 1),  // 步骤1: 创建一个信箱
      }
      saveOneIDScheduler.Schedule(task)           // 步骤2: 把任务投递到调度器
      return waitOneIDBatchResult(ctx, task.done) // 步骤3: 去信箱等结果
  }

  // 调度器（异步视角）
  func handleSaveOneIDBatch(ctx context.Context, groupKey string, tasks []batchscheduler.Task) error {
      // ... 批量调 API，拿到结果 ...
      for _, p := range payloads {
          p.done <- oneIDBatchResult{value: value} // 步骤4: 把结果放进信箱
          close(p.done)
      }
  }

  // 连接两者的桥梁
  func waitOneIDBatchResult(ctx context.Context, done <-chan oneIDBatchResult) (string, error) {
      select {
      case result := <-done:   // 步骤3在等这个：信箱里有没有东西？
          return result.value, result.err
      case <-ctx.Done():
          return "", ctx.Err()
      }
  }

  时序：

  调用方                          调度器(后台)
    |                               |
    Schedule(task) ──投递──→        |
    waitOneIDBatchResult            |
      select <-done ──阻塞等──      |
      |                             |
      |      (等几秒凑batch)         |
      |                             |
      |                  handleSaveOneIDBatch
      |                  调API拿结果
      |                  done <- result ──写入──→ done 有值了
      |                             |
      ←── select 解除阻塞，拿到值     |
    返回 openUID                    |

  done channel 是信箱：调用方创建信箱，投递任务后去信箱门口等。调度器处理完把结果放进信箱，调用方立刻收到。

  没有 channel 的话，调用方投递完任务就走了，调度器处理完也没法把结果送回去——异步就真的是纯异步，拿不到结果。

  channel 把"投递-等待-回传"串成一条线，对调用方来说就像调了一个普通函数：进去 → 等一会儿 → 拿到值出来。内部实际是异步聚合处理，但调用方感知不到。

退出场景

  
func (w *ConsumerHandler) drain(ctx context.Context) error {
    ctx = context.WithoutCancel(ctx)
    for {
        // 处理已经缓冲在 channel 里的消息，然后立刻退出
        select {
            case message, ok := <-w.Channel:
                if !ok || message == nil {
                   return nil
                }
                if err := w.Handle(ctx, message); err != nil {
                   		logger.Errorf(ctx, "handle_request_message %s %+v", message, err)
           				     }
            default:    //  default 本身没错，但它只有在 channel 有缓冲（≥1）
              return nil
        } 
    }
}

最常见：通道暂时为空（触发 default）

当 case 无法立即读取到数据时，select 会立刻命中 default 分支，执行 return nil，循环直接结束

通道被关闭（ok == false）

当 w.Channel 被外部 close() 关闭，且缓冲区里的数据已被取完时，下一次循环的 case 会立即返回零值，且 ok == false，触发 if !ok，执行 return nil。

收到约定的结束标记（message == nil）

如果你的业务逻辑约定往通道里发送 nil 指针代表“结束信号”，那么 if !ok || message == nil 条件成立，执行 return nil。

select随机

  
// 非阻塞地
select {
    case <-w.stopCh: // 已停 → 立即拒绝，绝不投递
    return context.Canceled
    default: // 没停 → 放行，继续
    }

// select随机，没有非阻塞地, 可能把消息投进死channel
select {
    case <-ctx.Done():
    return ctx.Err()
    case <-w.stopCh:
    return context.Canceled
    case w.Channel <- cache:
    return nil
}

退出

for message := range w.Channel 的语义是:

先从 channel 里取出所有已缓冲、还没被接收的消息,每条都会正常进入循环体,走 w.Handle(ctx, message)。
只有当 channel 既已关闭、又被取空了,range 才会停止循环。
close 不会让 range 立刻退出

  
// close(w.Channel) 时
for message := range w.Channel {
    // 这里只会执行"剩余的真实消息"
    // close 不会让 message 变成 nil 走进来
    if message == nil {   // ← 这个分支对 close 来说是死代码
        return nil
    }
}
return nil   // ← close + 取空后,从这里正常返回

//真正能"中断" Handle 的只有这些
/**
ctx 被 cancel —— 前提是 w.Handle 内部主动检查 ctx(比如 select 监听 ctx.Done())。close(w.Channel) 本身不会 cancel ctx。
Handle 内部 panic —— 会被最外层的 defer recover 接住,但那是它自己崩了,和 close 无关。
Handle 自己返回 —— 正常路径。
**/


// 

是的，完全允许多次调用，且绝对安全。

context.WithCancel 返回的 cancelFunc 在设计上就是幂等（Idempotent）的

取消上下文&关闭通道

. `cancelConsumer()` 的幂等性（安全）

结论：允许多次调用，绝对安全。
得益于 context 包的设计，第二次及之后的调用是空操作（No-op）。
实战价值：即使你在业务代码中因报错提前手动调用了 cancelConsumer()，函数返回时 defer 中的再次调用也不会引发 Panic，非常适合做“兜底清理”。

`close(w.Channel)` 的非幂等性（高风险）

结论：绝对不允许多次调用。 重复 close 会直接引发 panic: close of closed channel。
对比：cancelConsumer() 可以随便调用，但 close(ch) 不行。
双重保障：如果你无法保证只有一个地方调用 close（比如多个 defer 都想收尾），必须使用 sync.Once 包裹 close 操作。

必须使用

  
closeOnce sync.Once
func (w *PulsarConsumer) CloseChannel() {
        w.closeOnce.Do(func() {
            close(w.Channel)
        })
}

案例

例子

多路等待（multiplex）

  
func fastest(ctx context.Context, req Request) (Response, error) {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()          // 1. 确保函数退出时所有 goroutine 都被打断

	type item struct {
		resp Response
		err  error
	}
	ch := make(chan item, 2) // 缓冲 2，防止写端阻塞

	// 2. 并发请求两个副本
	go func() {
		resp, err := callReplica1(ctx, req)
		select {
		case ch <- item{resp, err}:
		case <-ctx.Done(): // 如果已经取消，直接退出
		}
	}()

	go func() {
		resp, err := callReplica2(ctx, req)
		select {
		case ch <- item{resp, err}:
		case <-ctx.Done():
		}
	}()

   // select阻塞, 没加循环 
	// 3. 只要第一个回来就返回，cancel 会通知另一个 goroutine 退出
	select {
	case it := <-ch:
		  return it.resp, it.err
	case <-ctx.Done():
		  return Response{}, ctx.Err()
	}
}

例子

  
func (w *ConsumerHandler) Run(ctx context.Context) error {
	defer func() {
		if err := recover(); err != nil {
			logger.Errorf(ctx, "handle_request_message recover err: %v stack:%v", err, string(debug.Stack()))
		}
	}()
	logger.Infof(ctx, "start handle_request_message %s", w.Name())
	for {
		select {
		case <-ctx.Done():
			return nil
		case message, ok := <-w.Channel:
			if !ok || message == nil { //!ok = 通道被关闭（外部 close(w.Channel)）；message == nil = 有人往通道里塞了 nil（常见用作停止信号）。两者都触发 return nil 退出。
				return nil
			}
			if err := w.Handle(ctx, message); err != nil {
				logger.Errorf(ctx, "handle_request_message %s %+v", message, err)
			}
		}
	}
}

取消goroutine

方式1 取消信号

案例1

  
done := make(chan struct{})
go func() {
    defer close(done)   // 干完活就 close
    doSomething()
}()

<-done  // 阻塞，直到 goroutine 关闭 done，表示完成

案例2

假设你有一个主 goroutine 和一个工作 goroutine，主 goroutine 需要通知工作 goroutine 停止处理，并等待工作 goroutine 确认停止。 主 goroutine

  
package main

import (
    "fmt"
    "time"
)

func main() {
    closeProcessCh := make(chan struct{})
    closeProcessAckCh := make(chan struct{})

    // 启动工作 goroutine
    go worker(closeProcessCh, closeProcessAckCh)

    // 模拟主 goroutine 的其他操作
    time.Sleep(2 * time.Second)

    // 通知工作 goroutine 停止处理
    closeProcessCh <- struct{}{}     // ① 发送停止信号（阻塞直到worker接收）//写
    <-closeProcessAckCh           // ② 等待确认（阻塞直到worker发送）   //读

    fmt.Println("工作 goroutine 已经停止")
}

工作 goroutine

  
func worker(closeProcessCh chan struct{}, closeProcessAckCh chan struct{}) {
    for {
        select {
        case <-closeProcessCh:
            // 收到停止信号，进行清理操作
            fmt.Println("工作 goroutine 收到停止信号，开始清理...")
            time.Sleep(1 * time.Second) // 模拟清理操作
            fmt.Println("工作 goroutine 清理完成")

            // 发送确认信号
            closeProcessAckCh <- struct{}{}
            return
        default:
            // 继续处理任务
            fmt.Println("工作 goroutine 正在处理任务...")
            time.Sleep(500 * time.Millisecond)
        }
    }
}

示例

  
	succPlatGoodsIds := make(map[string]bool)
	failedPlatGoodsIds := make(map[string]bool)	
	syncM := sync.Mutex{}
	runner := gorunner.NewGoRunner()
	runner.GoRun(ctx, func() error {
		for iterResult := range onsaleCh {
			if iterResult.Err != nil {
				abort = true
				return err
			}
			tempV := iterResult.Spu
			_, err := w.container.Spu().AddSpu(ctx, &tempV)
			if err != nil {
				syncM.Lock()
				failedPlatGoodsIds[tempV.PlatGoodsID] = true
				syncM.Unlock()
				continue
			} else {
				logger.Debugf(ctx, "add spu success, goods is %v", tempV.PlatGoodsID)
				syncM.Lock()
				succPlatGoodsIds[tempV.PlatGoodsID] = true
				actualTotal++
				syncM.Unlock()
			}
		}
		return nil
	})

	closeProcessCh := make(chan struct{})
	closeProcessAckCh := make(chan struct{})
	runner.GoRun(ctx, func() error {
		defer func() {
			closeProcessAckCh <- struct{}{}
		}()

		t := time.NewTicker(time.Second * 5)
		defer t.Stop()

		for {
			select {
			case <-t.C:
				syncM.Lock()
				succ := int64(len(succPlatGoodsIds))
				failed := int64(len(failedPlatGoodsIds))
				syncM.Unlock()

				fixTotal := max(total, succ+failed)
				logger.WithFields(taskFields).WithFields(logger.Fields{
					"trueTotal": actualTotal,
					"total":     fixTotal,
					"succeed":   succ,
					"failed":    failed,
				}).Info(ctx, "update task process")
				err = w.TaskProcessReport(fixTotal, succ, failed)
				if err != nil {
					logger.WithFields(taskFields).WithError(err).Error(ctx, "report process failed")
				}
			case <-closeProcessCh:
				return nil
			}
		}
	}, gorunner.WithSkipWait())
	runner.Wait()
	closeProcessCh <- struct{}{}
	<-closeProcessAckCh

方式2 取消上下文

  
onsaleCh 通道是在多个 fetchAndSendSpuItems 调用之间共享的。这意味着每个 fetchAndSendSpuItems 函数都会尝试向同一个 onsaleCh 通道发送数据。尝试关闭一个已经关闭的通道好像就会触发报错

示例

  
func (a *defaultApi) IterOnsaleGoodsByPage(ctx context.Context, work *Worker, pf platform.TPlatform, shopID string, initialPageNo int64, pageSize int64, total int64) (<-chan IterGoodsResult, error) {
	onsaleCh := make(chan IterGoodsResult, 150)
	subCtx, cancel := context.WithCancel(ctx)   // 取消上下文
	cancelFunc := func() { cancel() }
	runner := gorunner.NewGoRunner()
	var limitCount int64 = 10000

	if total >= limitCount {
		startTime := int64(1575043200)
		endTime := common.LocalTimeNow().Unix()
		isSubdivided, isAbort := efficientSubdivideTime(subCtx, work, pf, shopID, limitCount, initialPageNo, pageSize, startTime, endTime, onsaleCh, &runner, cancelFunc)
		if isAbort {
			cancelFunc()
			return onsaleCh, fmt.Errorf("unable to subdivide time range for shop %s", shopID)
		}
		if !isSubdivided {
			cancelFunc()
			return onsaleCh, fmt.Errorf("subdivision did not return any valid ranges for shop %s", shopID)
		}
	} else if total > 0 {
		runner.GoRun(ctx, func() error {
			return fetchAndSendSpuItems(subCtx, work, pf, shopID, initialPageNo, pageSize, nil, nil, onsaleCh, cancelFunc)
		})
	}

	// 后台等待所有任务完成
	gorunner.GoRun(ctx, func() error {
		defer func() {
			defer cancelFunc()  // 取消
			defer close(onsaleCh) // 完成后关闭 onsaleCh 通道
			logger.Infof(ctx, "IterOnsaleGoodsByPage all goroutines completed for shop %s", shopID)
		}()

		select {
		case <-ctx.Done():
			logger.Warnf(ctx, "Operation cancelled for shop %s", shopID)
			return ctx.Err()
		default:
			runner.Wait()
		}
		return nil
	})

	return onsaleCh, nil
}

// 获取SPU列表并发送到通道  //返回一个通道的函数
func fetchAndSendSpuItems(ctx context.Context, work *Worker, pf platform.TPlatform, shopID string, initialPageNo int64, pageSize int64, beginTime *int64, endTime *int64, onsaleCh chan<- IterGoodsResult, cancelFunc func()) error {
	filter := &spu.SpuQuery{
		ShopId:   shopID,
		Page:     initialPageNo,
		PageSize: pageSize,
	}

	if beginTime != nil && endTime != nil {
		filter.TimeFilters = []query.TimeFilter{
			{
				Start:    strconv.FormatInt(*beginTime, 10),
				End:      strconv.FormatInt(*endTime, 10),
				TimeName: "CreateTime",
			},
		}
	}

	for {
		select {
		case <-ctx.Done():
			onsaleCh <- IterGoodsResult{Err: ctx.Err()}
			return ctx.Err()
		default:
		}

		const maxRetries = 3
		retryBackoff := time.Second
		var spuList []spu.Spu
		var err error
		var i int
		// 重试
		for i = 0; i < maxRetries; i++ {
			spuList, _, err = work.container.Sdk(pf).GetSpuList(ctx, filter)
			if err == nil {
				break
			}
			time.Sleep(retryBackoff)
			logger.Infof(ctx, "failed to get spu list by page, shop_id: %v, page: %v, page_size: %v, err: %v, retrying (%d/%d)",
				filter.ShopId, filter.Page, filter.PageSize, err, i+1, maxRetries)
			retryBackoff *= 2 // 指数退避
		}
		if err != nil {
			logger.Errorf(ctx, "failed to get spu list by page, shop_id: %v, page: %v, page_size: %v, err: %v, retrying (%d/%d)",
				filter.ShopId, filter.Page, filter.PageSize, err, i+1, maxRetries)
			onsaleCh <- IterGoodsResult{Err: err}
			cancelFunc() // 取消所有goroutine的fetchAndSendSpuItems()
			return err
		}

		if len(spuList) == 0 {
			return nil
		}

		for _, spuItem := range spuList {
			s := spuItem
			onsaleCh <- IterGoodsResult{Spu: s, Err: nil}
		}

		filter.Page += 1
		if len(spuList) > 0 {
			time.Sleep(time.Second * 5)
		}
	}
}

// 细分时间区间
func efficientSubdivideTime(ctx context.Context, work *Worker, pf platform.TPlatform, shopID string, limitCount, initialPageNo, pageSize,
	startTime, endTime int64, onsaleCh chan<- IterGoodsResult, runner *gorunner.GoRunner, cancelFunc func()) (bool, bool) {
	if endTime <= startTime {
		return true, false // 参数1: true 表示区间有效; 参数2: true 表示是否终止
	}

	count, err := Instance(pf).GetOnsaleGoodsCountByTimestamp(ctx, work, pf, shopID, 1, 1, startTime, endTime)
	if err != nil {
		logger.Errorf(ctx, "failed to get spu list count by page, shop_id: %v, err: %v", shopID, err)
		onsaleCh <- IterGoodsResult{Err: err}
		return false, true
	}

	if count == 0 {
		return true, false
	}

	if count > 0 && count < limitCount {
		logger.Infof(ctx, "Efficient subdivision: startTime:%s, endTime:%s, count:%d", common.TimestampToStringTime(startTime), common.TimestampToStringTime(endTime), count)
		(*runner).GoRun(ctx, func() error {
			return fetchAndSendSpuItems(ctx, work, pf, shopID, initialPageNo, pageSize, &startTime, &endTime, onsaleCh, cancelFunc)
		})
		return true, false
	}

	if count >= limitCount {
		midTime := (startTime + endTime) / 2
		isLeftValid, isLeftAbort := efficientSubdivideTime(ctx, work, pf, shopID, limitCount, initialPageNo, pageSize, startTime, midTime, onsaleCh, runner, cancelFunc)
		if isLeftAbort {
			return false, true
		}

		isRightValid, isRightAbort := efficientSubdivideTime(ctx, work, pf, shopID, limitCount, initialPageNo, pageSize, midTime, endTime, onsaleCh, runner, cancelFunc)
		if isRightAbort {
			return false, true
		}

		return isLeftValid || isRightValid, false
	}

	return false, false
}

具体调用见: 方式1 取消信号

效果: 调用瞬间, 会打印细分出来的时间区间, 然后后台去执行耗时操作(fetchAndSendSpuItems)

为什么”先消费堆积”只属于 close 语义

你前面理解的那个”先取空缓冲”的行为，前提是 close(ch)：

复制

ch := make(chan item, 2)
ch <- it1
ch <- it2
close(ch)  // ← 关键：close 了

m, ok := <-ch  // ok=true, 拿到 it1（缓冲值）
m, ok = <-ch   // ok=true, 拿到 it2（缓冲值）
m, ok = <-ch   // ok=false, 取空了

这是 Go 通道的确定性语义：close 后，接收方会先把缓冲区里的值全部取完（期间 ok 一直是 true），取空之后才返回 ok=false。所以这里确实”先消费堆积”。

ctx 取消是另一回事

回到你的代码：

复制

for {
    select {
    case <-ctx.Done():          // ① ctx 取消通道
        return nil
    case message, ok := <-w.Channel:  // ② 消息通道
        if !ok || message == nil {
            return nil
        }
        w.Handle(ctx, message)
    }
}

这里有两个相互独立的 channel：

ctx.Done()：一旦取消/超时，它被关闭，从此永远处于”就绪”状态；
w.Channel：缓冲区里堆着消息时，它也”就绪”。

当 ctx.Done() 已关闭、同时 w.Channel 缓冲里还有消息时，select 的两个 case 都就绪。而 Go 的 select 在多个 case 就绪时是随机挑选的。

所以取消后的实际行为是：

轮次	可能结果
第 1 轮	随机：50% 走 `ctx.Done()` 直接退出；50% 拿到一条堆积消息处理
第 2 轮（若上轮处理了）	再次随机……
…	每次都随机，可能随时退出

堆积的 2 条消息可能一条都没处理，也可能处理 1 条、2 条，完全随机，没有顺序保证。 一旦某轮 select 选中了 ctx.Done()，剩下的堆积消息就被丢弃、永远不处理。

对比总结

触发方式	是否先消费堆积	原因
`close(w.Channel)`（`!ok`）	是，确定性	close 语义保证先取空缓冲，`ok` 才变 false
ctx 取消 / 超时	否，随机	`ctx.Done()` 与 `w.Channel` 在 select 里随机竞争

如果你确实想”取消前先把堆积处理完”

现在的代码做不到。需要改结构，比如：

复制

// 取消后：先非阻塞地 drain 掉缓冲里的消息，再退出
case <-ctx.Done():
    for {
        select {
        case message := <-w.Channel:
            w.Handle(ctx, message)  // 尽量处理完堆积
        default:
            return nil  // 没有堆积了，退出
        }
    }

不过要注意：这种”尽量 drain”仍然无法保证处理完所有堆积（处理过程中随时可能有新消息进入，且 ctx 已取消时继续处理是否合理需要业务判断）。是否要这样做，取决于你希望”优雅退出”到什么程度

GO, 异步

通道

内存泄露

阻塞判断

带缓存和无缓存

写法

搭配select

基础

主动取消（context.Done）

多路等待（multiplex）

非阻塞收发（带 default）

退出场景

select随机

退出

取消上下文&关闭通道

. cancelConsumer() 的幂等性（安全）

close(w.Channel) 的非幂等性（高风险）

案例

例子

例子

取消goroutine

方式1 取消信号

案例1

案例2

方式2 取消上下文

为什么”先消费堆积”只属于 close 语义

ctx 取消是另一回事

对比总结

如果你确实想”取消前先把堆积处理完”

. `cancelConsumer()` 的幂等性（安全）

`close(w.Channel)` 的非幂等性（高风险）