【问题标题】:How to write to two different csv files concurrently in Go?如何在 Go 中同时写入两个不同的 csv 文件?
【发布时间】:2021-12-16 10:43:05
【问题描述】:

我创建了一个最小的可复制示例

package main

import (
    "encoding/csv"
    "fmt"
    "os"
    "strconv"
    "sync/atomic"
    "time"
)

var (
    csvOnePath = "test.csv"
    csvTwoPath = "test_two.csv"
)

type A struct {
    Running     int32 // used atomically
    QuitChan    chan struct{}
}

func NewA() *A {
    return &A{
        QuitChan: make(chan struct{}),
    }
}

func (a *A) Start() error {
    if ok := atomic.CompareAndSwapInt32(&a.Running, 0, 1); !ok {
        return fmt.Errorf("Cannot start service A: service already started")
    }
    go a.record()
    return nil
}

func (a *A) Stop() error {
    if ok := atomic.CompareAndSwapInt32(&a.Running, 1, 0); !ok {
        return fmt.Errorf("Cannot stop service A: service already stopped")
    }
    close(a.QuitChan)
    return nil
}

func (a *A) record() {
    //file_one, err := os.OpenFile(csvOnePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0755)
    file_one, err := os.Create(csvOnePath)
    if err != nil {
        fmt.Println(err)
        return
    }
    writer := csv.NewWriter(file_one)
    // writer, closeFileFunc, err := NewCsvWriter(csvOnePath)
    if err != nil {
        fmt.Println(err)
        return
    }
    header := []string{"this", "is", "a", "test"}
    err = writer.Write(header)
    if err != nil {
        fmt.Println(err)
        return
    }
    ticker := time.NewTicker(10*time.Second)
    for {
        select {
        case t := <-ticker.C:
            err = writer.Write([]string{fmt.Sprintf("%2d:%2d:%2d", t.Hour(), t.Minute(), t.Second())})
            if err != nil {
                fmt.Println(err)
                a.QuitChan <- struct{}{}
            }
        case <-a.QuitChan:
            ticker.Stop()
            writer.Flush()
            file_one.Close()
            fmt.Println("Stopped recording.")
            break
        }
    }
}

type B struct {
    Running     int32 // used atomically
    QuitChan    chan struct{}
}

func NewB() *B {
    return &B{
        QuitChan: make(chan struct{}),
    }
}

func (b *B) Start() error {
    if ok := atomic.CompareAndSwapInt32(&b.Running, 0, 1); !ok {
        return fmt.Errorf("Cannot start service B: service already started")
    }
    go b.record()
    return nil
}

func (b *B) Stop() error {
    if ok := atomic.CompareAndSwapInt32(&b.Running, 1, 0); !ok {
        return fmt.Errorf("Cannot stop service B: service already stopped")
    }
    close(b.QuitChan)
    return nil
}

func writeMsgToReport(report *csv.Writer, msg string) error {
    ct := time.Now()
    timestamp := fmt.Sprintf("%2d:%2d:%2d", ct.Hour(), ct.Minute(), ct.Second())
    return report.Write([]string{timestamp, msg})
}

func (b *B) record() {
    //file_two, err := os.OpenFile(csvTwoPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0666)
    file_two, err := os.Create(csvTwoPath)
    if err != nil {
        fmt.Println(err)
        return
    }
    writer := csv.NewWriter(file_two)
    //writer, closeFileFunc, err := NewCsvWriter(csvTwoPath)
    if err != nil {
        fmt.Println(err)
        return
    }
    header := []string{"this", "is", "a", "second", "test"}
    err = writer.Write(header)
    if err != nil {
        fmt.Println(err)
        return
    }
    ticker := time.NewTicker(1*time.Second)
    ticks := 0
    for {
        select {
        case <-ticker.C:
            if ticks % 15 == 0 {
                err = writeMsgToReport(writer, "YEET "+strconv.Itoa(ticks))
                if err != nil {
                    fmt.Println(err)
                    b.QuitChan <- struct{}{}
                }
            }
            ticks++
        case <-b.QuitChan:
            ticker.Stop()
            writer.Flush()
            file_two.Close()
            fmt.Println("Stopped recording.")
            break
        }
    }
}

func main() {
    serviceA := NewA()
    err := serviceA.Start()
    if err != nil {
        fmt.Println(err)
        return
    }
    defer serviceA.Stop()
    serviceB := NewB()
    err = serviceB.Start()
    if err != nil {
        fmt.Println(err)
        return
    }
    defer serviceB.Stop()
    time.Sleep(600*time.Second)
}

基本上,我有两个不同的服务,它们在两个不同的 goroutine 中运行 record 方法。它们各自在不同时间创建并写入不同的 csv 文件。当我运行它时,会创建 csv 文件但永远不会有数据。运行此程序时不会引发任何错误。我读到我应该使用我已经实现的互斥锁,但这也没有奏效。我应该在这里做什么?

【问题讨论】:

  • 我简要浏览了代码并注意到closeFileFunc()writer.Flush() 之前被调用。是否应该交换订单?看起来不需要 CsvWriter 中的互斥锁,因为没有对给定 CsvWriter 的并发访问。如果这是正确的,那么直接从记录方法中使用 enconding/csv 包来简化代码。
  • 我将颠倒关闭和刷新的顺序,移除互斥体并报告回来。谢谢!
  • 我不确定通道方法是不是最好的,尽管有可能。你可以使用 WaitGroup 来完成你想要的,通过添加 2 并简单地阻塞直到它们完成。
  • 我需要通道方法才能工作。在实际项目中,我正在以不同的时间间隔将来自两个不同来源的数据实时写入两个不同的文件。
  • 根据the spec,当main() 完成“它不会等待其他(非主)goroutines 完成”时,您的程序将退出。 (example)。这意味着您的goroutines可能在刷新/关闭文件之前被杀死。有几种方法可以解决这个问题; WaitGroup 是显而易见的。

标签: csv go concurrency goroutine


【解决方案1】:

如 cmets 中所述,程序将在 main() 完成时退出; the spec 声明“它不会等待其他(非主)goroutines 完成。”。

这意味着您的 go 例程不太可能处理关闭文件的代码,这意味着可能不会写入缓冲数据。

我在the playground 中创建了您的应用程序的简化版本来演示这一点。

有很多方法可以解决这个问题,但最简单的可能是添加一个WaitGroup,这样您的应用程序就可以在终止之前等待 go 例程退出。

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2022-01-12
    • 1970-01-01
    • 2020-11-30
    • 1970-01-01
    • 2018-05-09
    • 1970-01-01
    • 2017-06-24
    相关资源
    最近更新 更多