go - 在时间间隔和 channel 长度之间选择

Question

我来这里是为了找出执行后续任务的最惯用的方法。

任务:

将数据从 channel 写入文件。

问题:

我有一个 channel ch := make(chan int, 100)

我需要从 channel 读取并将我从 channel 读取的值写入文件。我的问题基本上是我该怎么做，因为

1. 如果 channel ch 已满，立即写入值
2. 如果 channel ch 未满，则每 5 秒写入一次。

所以本质上，数据需要至少每 5s 写入文件(假设数据至少每 5s 将被填充到 channel 中)

使用 select、for 和 range 完成上述任务的最佳方法是什么？

谢谢!

Answer 1

没有“ channel 缓冲区已满”这样的“事件”，因此您无法检测到[*]。这意味着您无法仅使用 1 个 channel 以惯用的方式解决您的语言原语问题。

^{[*] 不完全正确:当发送 在 channel 上，但这需要发送者的逻辑，并重复发送尝试。}

我会使用另一个 channel ，当值在其上发送时，我会从该 channel 接收值，然后“重定向”，将值存储在另一个 channel 中，如您所述，该 channel 的缓冲区为 100。在每次重定向时，您可以检查内部 channel 的缓冲区是否已满，如果是，则立即写入。如果没有，继续使用 select 语句监视“传入” channel 和计时器 channel ，如果计时器触发，则执行“常规”写入。

您可以使用 len(chInternal) 检查 chInternal channel 中有多少元素，并使用 cap(chInternal) 检查其容量。请注意，这是“安全的”，因为我们是唯一处理 chInternal channel 的 goroutine。如果有多个 goroutine，则 len(chInternal) 返回的值在我们将它用于某些东西(例如比较它)时可能已经过时。

在此解决方案中，chInternal(如其名称所示)仅供内部使用。其他人应该只在 ch 上发送值。请注意，ch 可能是也可能不是缓冲 channel ，解决方案适用于这两种情况。但是，如果您还为 ch 提供一些缓冲区，则可能会提高效率(这样发件人被阻止的可能性就会降低)。

var (
    chInternal = make(chan int, 100)
    ch         = make(chan int) // You may (should) make this a buffered channel too
)

func main() {
    delay := time.Second * 5
    timer := time.NewTimer(delay)
    for {
        select {
        case v := <-ch:
            chInternal <- v
            if len(chInternal) == cap(chInternal) {
                doWrite() // Buffer is full, we need to write immediately
                timer.Reset(delay)
            }
        case <-timer.C:
            doWrite() // "Regular" write: 5 seconds have passed since last write
            timer.Reset(delay)
        }
    }
}

如果发生立即写入(由于“缓冲区已满”的情况)，此解决方案将在此之后的 5 秒内安排下一次“常规”写入。如果您不想这样，并且希望 5 秒的常规写入独立于立即写入，那么只需不要在立即写入后重置计时器即可。

doWrite()的实现可能如下:

var f *os.File // Make sure to open file for writing

func doWrite() {
    for {
        select {
        case v := <-chInternal:
            fmt.Fprintf(f, "%d ", v) // Write v to the file
        default: // Stop when no more values in chInternal
            return
        }
    }
}

我们不能使用 for ... range 因为它只在 channel 关闭时返回，但我们的 chInternal channel 没有关闭。因此，我们将 select 与 default 一起使用，这样当 chInternal 的缓冲区中没有更多值时，我们就返回。

改进

使用 slice 而不是第二个 channel

由于 chInternal channel 仅供我们使用，并且仅在单个 goroutine 上使用，我们也可以选择使用单个 []int slice 而不是 channel (读/写 slice 比 channel 快得多)。

仅显示不同/更改的部分，它可能看起来像这样:

var (
    buf = make([]int, 0, 100)
)

func main() {
    // ...

    for {
        select {
        case v := <-ch:
            buf = append(buf, v)
            if len(buf) == cap(buf) {
            // ...
    }
}

func doWrite() {
    for _, v := range buf {
        fmt.Fprintf(f, "%d ", v) // Write v to the file
    }
    buf = buf[:0] // "Clear" the buffer
}

有多个协程

如果我们坚持为 chInternal 留一个 channel ，则可以在另一个 goroutine 上调用 doWrite() 函数以不阻塞另一个 goroutine，例如执行 doWrite()。由于要写入的数据是从 channel (chInternal) 读取的，因此不需要进一步同步。