为何带超时的Context中select语句里的context.Done()从未触发？

以下是我用于读取Google Pub/Sub消息的代码：

func (p *PubSubSource) Read(_ context.Context, readRequest sourcesdk.ReadRequest, messageCh chan<- sourcesdk.Message) {
    if len(p.messages) > 0 {
        return
    }
    log.Println("Request timeout", readRequest.TimeOut())
    ctx, cancelFunc := context.WithTimeout(context.Background(), readRequest.TimeOut())
    defer cancelFunc()
    cctx, cancel := context.WithCancel(ctx)
    receiveMesg := make(chan *pubsub.Message)
    p.subscription.ReceiveSettings.MaxOutstandingMessages = int(readRequest.Count())
    go p.subscription.Receive(cctx, func(_ context.Context, msg *pubsub.Message) {
        fmt.Printf("Got message------------------: %s\n", string(msg.Data))
        receiveMesg <- msg

    })
    for {
        select {
        case <-ctx.Done():
            log.Println("Timeout done ------------************************")
            cancel()
            return

        case msg := <-receiveMesg:
            log.Println("Executing Loop--------")

            p.lock.Lock()
            messageCh <- sourcesdk.NewMessage(
                msg.Data,
                sourcesdk.NewOffset([]byte(msg.ID), "0"),
                msg.PublishTime,
            )
            p.messages[msg.ID] = msg
            p.lock.Unlock()

        default:
            continue

        }

    }
}

问题在于case <-ctx.Done()从未执行，我希望超时后退出父函数，但ctx.Done()从未被触发，请问这是哪里出错了？有没有更好的实现方式？

问题原因 #

1. select的default分支抢占了检测机会：你的select里加了default: continue，这会让select变成非阻塞轮询模式。当没有消息、ctx也没超时的时候，会一直循环走default分支；哪怕ctx超时后Done()通道就绪，高频率的轮询也会大概率跳过Done()的case，导致超时信号无法被及时捕获。

2. 无缓冲通道导致回调阻塞：receiveMesg是无缓冲通道，当Pub/Sub回调发送消息时，如果主循环没及时接收，回调会被卡住，可能影响Receive函数对上下文取消信号的响应，间接导致流程异常。

修复方案 #

方案1：移除default分支，优化通道与资源清理

去掉default分支让select阻塞等待事件，同时给通道加缓冲避免回调阻塞，确保超时信号能被及时处理：

func (p *PubSubSource) Read(_ context.Context, readRequest sourcesdk.ReadRequest, messageCh chan<- sourcesdk.Message) {
    if len(p.messages) > 0 {
        return
    }
    log.Println("Request timeout", readRequest.TimeOut())
    ctx, cancelFunc := context.WithTimeout(context.Background(), readRequest.TimeOut())
    defer cancelFunc()
    cctx, cancel := context.WithCancel(ctx)
    defer cancel() // 退出时确保取消子上下文
    
    // 给通道加缓冲，大小和请求消息数一致，避免回调阻塞
    receiveMesg := make(chan *pubsub.Message, int(readRequest.Count()))
    p.subscription.ReceiveSettings.MaxOutstandingMessages = int(readRequest.Count())
    
    go func() {
        defer close(receiveMesg) // goroutine退出时关闭通道，让主循环能检测到
        p.subscription.Receive(cctx, func(_ context.Context, msg *pubsub.Message) {
            fmt.Printf("Got message------------------: %s\n", string(msg.Data))
            receiveMesg <- msg
        })
    }()

    for {
        select {
        case <-ctx.Done():
            log.Println("Timeout done ------------************************")
            return
        case msg, ok := <-receiveMesg:
            if !ok {
                // 通道关闭，退出循环
                log.Println("Receive channel closed")
                return
            }
            log.Println("Executing Loop--------")

            p.lock.Lock()
            messageCh <- sourcesdk.NewMessage(
                msg.Data,
                sourcesdk.NewOffset([]byte(msg.ID), "0"),
                msg.PublishTime,
            )
            p.messages[msg.ID] = msg
            p.lock.Unlock()
        }
    }
}

方案2：简化上下文使用，回调内检测取消信号

直接将超时上下文传给Receive，并在回调里检查上下文状态，避免不必要的子上下文，同时处理超时后的消息：

func (p *PubSubSource) Read(_ context.Context, readRequest sourcesdk.ReadRequest, messageCh chan<- sourcesdk.Message) {
    if len(p.messages) > 0 {
        return
    }
    log.Println("Request timeout", readRequest.TimeOut())
    ctx, cancelFunc := context.WithTimeout(context.Background(), readRequest.TimeOut())
    defer cancelFunc()
    
    receiveMesg := make(chan *pubsub.Message, int(readRequest.Count()))
    p.subscription.ReceiveSettings.MaxOutstandingMessages = int(readRequest.Count())
    
    go func() {
        defer close(receiveMesg)
        p.subscription.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
            // 先检查上下文是否已取消，避免处理超时后的消息
            select {
            case <-ctx.Done():
                msg.Nack() // 未确认消息，让Pub/Sub重新投递
                return
            default:
                fmt.Printf("Got message------------------: %s\n", string(msg.Data))
                receiveMesg <- msg
            }
        })
    }()

    for {
        select {
        case <-ctx.Done():
            log.Println("Timeout done ------------************************")
            return
        case msg, ok := <-receiveMesg:
            if !ok {
                log.Println("Receive channel closed")
                return
            }
            log.Println("Executing Loop--------")

            p.lock.Lock()
            messageCh <- sourcesdk.NewMessage(
                msg.Data,
                sourcesdk.NewOffset([]byte(msg.ID), "0"),
                msg.PublishTime,
            )
            p.messages[msg.ID] = msg
            p.lock.Unlock()
        }
    }
}

关键优化点 #

• 移除default分支，让select阻塞等待事件，确保ctx.Done()信号能被及时捕获。
• 给receiveMesg加缓冲，避免Pub/Sub回调被阻塞。
• 在goroutine中关闭通道，让主循环能检测到通道关闭并正常退出。
• 回调内检查上下文状态，避免处理超时后的无效消息，同时调用msg.Nack()保证消息不丢失。

内容的提问来源于stack exchange，提问作者gANDALF