我来这里是为了找出最惯用的方法来完成跟随任务。
任务：
将数据从通道写入文件。
问题：
我有一个频道ch := make(chan int, 100)
我需要从通道中读取数据并将从通道中读取的值写入文件。我的问题基本上是考虑到我该怎么做
ch已满，立即写入值ch未满，则每5s写入一次。 所以本质上，需要至少每 5s 将数据写入文件（假设数据将至少每 5s 填充到通道中）
使用select、for 和range 完成上述任务的最佳方式是什么？
谢谢！
【问题讨论】：
int 值，还是要写入频道的全部内容（清空其缓冲区）？
没有像“通道缓冲区已满”这样的“事件”，所以你无法检测到[*]。这意味着您无法惯用地解决仅使用 1 个通道的语言原语问题。
[*] 不完全正确：当在通道上 发送 时，您可以通过使用 select 和 default 情况来检测通道的缓冲区是否已满，但这需要来自发件人的逻辑，以及重复的发送尝试。
我会使用另一个通道，当值在其上发送时，我会从该通道接收，然后“重定向”，将值存储在另一个通道中，该通道具有 100 的缓冲区，如您所述。在每次重定向时，您可以检查内部通道的缓冲区是否已满，如果是，请立即写入。如果没有，继续使用select 语句监视“传入”通道和计时器通道，如果计时器触发，则执行“常规”写入。
您可以使用len(chInternal) 来检查chInternal 通道中有多少元素，并使用cap(chInternal) 来检查其容量。请注意，这是“安全的”，因为我们是唯一处理 chInternal 通道的 goroutine。如果有多个 goroutine，len(chInternal) 返回的值可能在我们使用它时（例如比较它）已经过时了。
在此解决方案中，chInternal（顾名思义）仅供内部使用。其他人只能在ch 上发送值。请注意，ch 可能是也可能不是缓冲通道，解决方案在这两种情况下都适用。但是，如果您也给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，因为它只在通道关闭时返回，但我们的chInternal 通道没有关闭。所以我们使用select 和default 的情况，所以当chInternal 的缓冲区中没有更多值时，我们返回。
由于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 一个频道，doWrite() 函数可能会在另一个 goroutine 上被调用而不阻塞另一个，例如go doWrite()。由于要写入的数据是从通道 (chInternal) 读取的，因此不需要进一步同步。
【讨论】：
ch，那么第一个case会执行5次（共10s）。但是我们需要每五秒写一次。
ch，是的，第一个 case 将每 2 秒执行一次，但写入不会发生只有在 @ 的缓冲区987654350@ 已满（请参阅if 语句），但不会满，因此第一次写入只会在timer 触发时的 5 秒后发生（应该如此）。
doWrite() 函数也会变得更简单。
如果只使用 5 秒写入，以提高文件写入性能，
您可以随时填写频道，
然后 writer goroutine 将该数据写入缓冲文件，
在不使用计时器的情况下查看这个非常简单且惯用的示例
只使用 for...range：
package main
import (
"bufio"
"fmt"
"os"
"sync"
)
var wg sync.WaitGroup
func WriteToFile(filename string, ch chan int) {
f, e := os.Create(filename)
if e != nil {
panic(e)
}
w := bufio.NewWriterSize(f, 4*1024*1024)
defer wg.Done()
defer f.Close()
defer w.Flush()
for v := range ch {
fmt.Fprintf(w, "%d ", v)
}
}
func main() {
ch := make(chan int, 100)
wg.Add(1)
go WriteToFile("file.txt", ch)
for i := 0; i < 500000; i++ {
ch <- i // do the job
}
close(ch) // Finish the job and close output file
wg.Wait()
}
并注意defers 订单。
如果写入时间为 5 秒，您可以添加一个间隔计时器，只是为了将此文件的缓冲区刷新到磁盘，如下所示：
package main
import (
"bufio"
"fmt"
"os"
"sync"
"time"
)
var wg sync.WaitGroup
func WriteToFile(filename string, ch chan int) {
f, e := os.Create(filename)
if e != nil {
panic(e)
}
w := bufio.NewWriterSize(f, 4*1024*1024)
ticker := time.NewTicker(5 * time.Second)
quit := make(chan struct{})
go func() {
for {
select {
case <-ticker.C:
if w.Buffered() > 0 {
fmt.Println(w.Buffered())
w.Flush()
}
case <-quit:
ticker.Stop()
return
}
}
}()
defer wg.Done()
defer f.Close()
defer w.Flush()
defer close(quit)
for v := range ch {
fmt.Fprintf(w, "%d ", v)
}
}
func main() {
ch := make(chan int, 100)
wg.Add(1)
go WriteToFile("file.txt", ch)
for i := 0; i < 25; i++ {
ch <- i // do the job
time.Sleep(500 * time.Millisecond)
}
close(ch) // Finish the job and close output file
wg.Wait()
}
这里我使用time.NewTicker(5 * time.Second)作为quit频道的间隔定时器，你可以使用time.AfterFunc()或time.Tick()或time.Sleep()。
进行了一些优化（移除退出通道）：
package main
import (
"bufio"
"fmt"
"os"
"sync"
"time"
)
var wg sync.WaitGroup
func WriteToFile(filename string, ch chan int) {
f, e := os.Create(filename)
if e != nil {
panic(e)
}
w := bufio.NewWriterSize(f, 4*1024*1024)
ticker := time.NewTicker(5 * time.Second)
defer wg.Done()
defer f.Close()
defer w.Flush()
for {
select {
case v, ok := <-ch:
if ok {
fmt.Fprintf(w, "%d ", v)
} else {
fmt.Println("done.")
ticker.Stop()
return
}
case <-ticker.C:
if w.Buffered() > 0 {
fmt.Println(w.Buffered())
w.Flush()
}
}
}
}
func main() {
ch := make(chan int, 100)
wg.Add(1)
go WriteToFile("file.txt", ch)
for i := 0; i < 25; i++ {
ch <- i // do the job
time.Sleep(500 * time.Millisecond)
}
close(ch) // Finish the job and close output file
wg.Wait()
}
我希望这会有所帮助。
【讨论】：