在没有接收器的情况下，是不是可以将数据打开的缓冲通道保留下来？

Posted 2023-03-14

【中文标题】在没有接收器的情况下，是不是可以将数据打开的缓冲通道保留下来？

【英文标题】：Is it OK to leave a buffered channel with data open when there is no receiver?在没有接收器的情况下，是否可以将数据打开的缓冲通道保留下来？

【发布时间】：2018-11-09 23:39:18

【问题描述】：

假设一个频道有 10 个发送者和一个接收者。发送方函数需要一些时间才能返回值。接收者只需要一个值（接收到的第一个值）来自通道，其他 9 个值不使用。接收者不需要等待剩下的 9 个值。这就是为什么我没有使用sync.WaitGroup。

我使用了缓冲通道，所以当接收器只取第一个时，缓冲通道中会有 9 个数据。我的问题是：

在没有接收器的情况下，是否可以让数据打开的缓冲通道保持打开状态？下面的示例代码是一个简化的，但如果程序是一个守护进程，它最终会被垃圾回收吗？

有没有更好的方法来处理这种情况？我尝试使用取消通道但失败了。我不确定context 是否适合这种情况。

示例代码：

package main

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

func main() 
    rand.Seed(time.Now().UnixNano())

    i, err := getRandomInt()
    if err != nil 
        fmt.Println(err)
     else 
        fmt.Println(i)
    

    fmt.Println("Waiting goroutines to be finished...")
    time.Sleep(2 * time.Second)


func getRandomInt() (int, error) 
    ch := make(chan int, 10)

    // 10 senders
    for i := 0; i < 10; i++ 
        go func(i int) 
            defer fmt.Printf("Goroutine #%d finished\n", i)
            fmt.Printf("Goroutine #%d started\n", i)

            data := heavyJob()
            ch <- data
            fmt.Printf("Goroutine #%d sent data %d to ch\n", i, data)
            return
        (i)
    

    // 1 receiver
    timeout := time.After(2000 * time.Millisecond)
    for 
        select 
        case value := <-ch:
            // uses only the value received first, the rest are discarded
            return value, nil
        case <-timeout:
            return -1, errors.New("Timeout")
        
    


// takes 1900~2900ms to finish
func heavyJob() int 
    r := rand.Intn(1000)
    time.Sleep(time.Duration(r+1900) * time.Millisecond)
    return r

Run on playground

【问题讨论】：

1.是和 2。不。一旦缓冲区被收集，其中的 9 个剩余项目就会被收集，所以如果没有其他限制，为什么还要使用更复杂的解决方案。顺便提一句。语言的名称是 Go。只有去。

【参考方案1】：

回答主要问题：

离开频道没问题，它会被垃圾回收。 这似乎是基于意见的，但对我来说，如果您创建具有 10 个空格的缓冲通道的唯一原因是让发送者 goroutine 可以退出；这感觉就像它会从重新设计中受益。还有其他（也许更好的）方法可以确保发送者 goroutine 可以关闭。

本质上，您正在创建工人数量和缓冲通道大小之间的隐式耦合。更改这两个数字中的一个，就会出现死锁/中断！ （附带说明一下，缓冲通道通常适用于消费者和生产者以相同的速率工作但没有稳定输出的情况。它是尖峰的，缓冲区平滑了波峰和波谷。）

考虑到这一点，我建议最好明确管理您不想要所有值的事实。

这是 getRandomInt() 函数的更新版本。请注意在顶部使用 defer 设置上下文取消，以及在发送时使用 select 语句。

func getRandomInt() (int, error) 
    ctx := context.Background() // creates a fresh, empty context
    ctx, cancel := context.WithCancel(ctx)
    defer cancel() // cancels the context when getRandomInt() returns

    ch := make(chan int)

    // 10 senders
    for i := 0; i < 10; i++ 
        go func(i int) 
            defer fmt.Printf("Goroutine #%d finished\n", i)
            fmt.Printf("Goroutine #%d started\n", i)

            data := heavyJob()

            // this select statement wil block until either this goroutine 
            // is the first to send, or the context is cancelled. In which case
            // another routine has already sent and it can discard it's values.
            select  
            case ch <- data:
                fmt.Printf("Goroutine #%d sent data %d to ch\n", i, data)
            case <-ctx.Done():
                fmt.Printf("Goroutine #%d did not send, context is cancelled, would have sent data %d to ch\n", i, data)
            
        (i)
    

    // 1 receiver
    timeout := time.After(2000 * time.Millisecond)
    select 
    case value := <-ch:
        // uses only the value received first, the rest are discarded
        return value, nil
    case <-timeout:
        return -1, errors.New("Timeout")
    

使用取消设置上下文意味着一旦调用cancel() 函数，上下文就会“完成”。这是一种告诉所有发送者 goroutine 不要等待发送的方法。

在发送时，select 语句会阻塞，直到上下文被cancel() 函数取消；或者接收器方法读取第一个值。

我还从通道中删除了缓冲，因为不再需要它了。

【讨论】：

@philipjkim，还请注意，现在将 ctx 传递给 HeavyJob 是微不足道的，这样他们就可以在第一个完成后中止。