How to write to two different csv files concurrently in Go?

huangapple
go评论126阅读模式
英文:

How to write to two different csv files concurrently in Go?

问题

我已经创建了一个最小可重现的示例。基本上，我有两个不同的服务，在两个不同的goroutine中运行record方法。它们分别在不同的时间创建并写入不同的CSV文件。当我运行这段代码时，CSV文件被创建了，但是没有数据。在运行过程中没有出现任何错误。我读到应该使用互斥锁，我已经实现了，但是也没有起作用。在这种情况下，我应该怎么做？

以下是修复问题的建议：

  1. 问题可能出在writer.Flush()方法没有被调用，导致数据没有被写入文件。在record方法的case <-a.QuitChan:和case <-b.QuitChan:分支中，添加writer.Flush()语句，确保数据被刷新到文件中。

  2. 为了避免并发访问文件的问题，可以使用互斥锁来保护对文件的写操作。在A和B结构体中分别添加一个互斥锁字段，并在写入文件之前使用Lock方法获取锁，在写入完成后使用Unlock方法释放锁。

修改后的代码如下：

  1. package main
  3. import (
  4. 	"encoding/csv"
  5. 	"fmt"
  6. 	"os"
  7. 	"strconv"
  8. 	"sync"
  9. 	"sync/atomic"
  10. 	"time"
  11. )
  13. var (
  14. 	csvOnePath = "test.csv"
  15. 	csvTwoPath = "test_two.csv"
  16. )
  18. type A struct {
  19. 	Running   int32 // used atomically
  20. 	QuitChan  chan struct{}
  21. 	mutex     sync.Mutex
  22. }
  24. func NewA() *A {
  25. 	return &A{
  26. 		QuitChan: make(chan struct{}),
  27. 	}
  28. }
  30. func (a *A) Start() error {
  31. 	if ok := atomic.CompareAndSwapInt32(&a.Running, 0, 1); !ok {
  32. 		return fmt.Errorf("Cannot start service A: service already started")
  33. 	}
  34. 	go a.record()
  35. 	return nil
  36. }
  38. func (a *A) Stop() error {
  39. 	if ok := atomic.CompareAndSwapInt32(&a.Running, 1, 0); !ok {
  40. 		return fmt.Errorf("Cannot stop service A: service already stopped")
  41. 	}
  42. 	close(a.QuitChan)
  43. 	return nil
  44. }
  46. func (a *A) record() {
  47. 	a.mutex.Lock()
  48. 	defer a.mutex.Unlock()
  50. 	file_one, err := os.Create(csvOnePath)
  51. 	if err != nil {
  52. 		fmt.Println(err)
  53. 		return
  54. 	}
  55. 	defer file_one.Close()
  57. 	writer := csv.NewWriter(file_one)
  58. 	defer writer.Flush()
  60. 	header := []string{"this", "is", "a", "test"}
  61. 	err = writer.Write(header)
  62. 	if err != nil {
  63. 		fmt.Println(err)
  64. 		return
  65. 	}
  67. 	ticker := time.NewTicker(10 * time.Second)
  68. 	for {
  69. 		select {
  70. 		case t := <-ticker.C:
  71. 			err = writer.Write([]string{fmt.Sprintf("%2d:%2d:%2d", t.Hour(), t.Minute(), t.Second())})
  72. 			if err != nil {
  73. 				fmt.Println(err)
  74. 				a.QuitChan <- struct{}{}
  75. 			}
  76. 		case <-a.QuitChan:
  77. 			ticker.Stop()
  78. 			fmt.Println("Stopped recording.")
  79. 			return
  80. 		}
  81. 	}
  82. }
  84. type B struct {
  85. 	Running   int32 // used atomically
  86. 	QuitChan  chan struct{}
  87. 	mutex     sync.Mutex
  88. }
  90. func NewB() *B {
  91. 	return &B{
  92. 		QuitChan: make(chan struct{}),
  93. 	}
  94. }
  96. func (b *B) Start() error {
  97. 	if ok := atomic.CompareAndSwapInt32(&b.Running, 0, 1); !ok {
  98. 		return fmt.Errorf("Cannot start service B: service already started")
  99. 	}
  100. 	go b.record()
  101. 	return nil
  102. }
  104. func (b *B) Stop() error {
  105. 	if ok := atomic.CompareAndSwapInt32(&b.Running, 1, 0); !ok {
  106. 		return fmt.Errorf("Cannot stop service B: service already stopped")
  107. 	}
  108. 	close(b.QuitChan)
  109. 	return nil
  110. }
  112. func (b *B) record() {
  113. 	b.mutex.Lock()
  114. 	defer b.mutex.Unlock()
  116. 	file_two, err := os.Create(csvTwoPath)
  117. 	if err != nil {
  118. 		fmt.Println(err)
  119. 		return
  120. 	}
  121. 	defer file_two.Close()
  123. 	writer := csv.NewWriter(file_two)
  124. 	defer writer.Flush()
  126. 	header := []string{"this", "is", "a", "second", "test"}
  127. 	err = writer.Write(header)
  128. 	if err != nil {
  129. 		fmt.Println(err)
  130. 		return
  131. 	}
  133. 	ticker := time.NewTicker(1 * time.Second)
  134. 	ticks := 0
  135. 	for {
  136. 		select {
  137. 		case <-ticker.C:
  138. 			if ticks%15 == 0 {
  139. 				err = writeMsgToReport(writer, "YEET "+strconv.Itoa(ticks))
  140. 				if err != nil {
  141. 					fmt.Println(err)
  142. 					b.QuitChan <- struct{}{}
  143. 				}
  144. 			}
  145. 			ticks++
  146. 		case <-b.QuitChan:
  147. 			ticker.Stop()
  148. 			fmt.Println("Stopped recording.")
  149. 			return
  150. 		}
  151. 	}
  152. }
  154. func writeMsgToReport(report *csv.Writer, msg string) error {
  155. 	ct := time.Now()
  156. 	timestamp := fmt.Sprintf("%2d:%2d:%2d", ct.Hour(), ct.Minute(), ct.Second())
  157. 	return report.Write([]string{timestamp, msg})
  158. }
  160. func main() {
  161. 	serviceA := NewA()
  162. 	err := serviceA.Start()
  163. 	if err != nil {
  164. 		fmt.Println(err)
  165. 		return
  166. 	}
  167. 	defer serviceA.Stop()
  169. 	serviceB := NewB()
  170. 	err = serviceB.Start()
  171. 	if err != nil {
  172. 		fmt.Println(err)
  173. 		return
  174. 	}
  175. 	defer serviceB.Stop()
  177. 	time.Sleep(600 * time.Second)
  178. }

请尝试使用上述修改后的代码，并确保在写入文件之后调用writer.Flush()方法。这样应该能够解决你的问题。

英文:

I've created a minimal reproduceable example

  1. package main
  2. import (
  3. &quot;encoding/csv&quot;
  4. &quot;fmt&quot;
  5. &quot;os&quot;
  6. &quot;strconv&quot;
  7. &quot;sync/atomic&quot;
  8. &quot;time&quot;
  9. )
  10. var (
  11. csvOnePath = &quot;test.csv&quot;
  12. csvTwoPath = &quot;test_two.csv&quot;
  13. )
  14. type A struct {
  15. Running 	int32 // used atomically
  16. QuitChan 	chan struct{}
  17. }
  18. func NewA() *A {
  19. return &amp;A{
  20. QuitChan: make(chan struct{}),
  21. }
  22. }
  23. func (a *A) Start() error {
  24. if ok := atomic.CompareAndSwapInt32(&amp;a.Running, 0, 1); !ok {
  25. return fmt.Errorf(&quot;Cannot start service A: service already started&quot;)
  26. }
  27. go a.record()
  28. return nil
  29. }
  30. func (a *A) Stop() error {
  31. if ok := atomic.CompareAndSwapInt32(&amp;a.Running, 1, 0); !ok {
  32. return fmt.Errorf(&quot;Cannot stop service A: service already stopped&quot;)
  33. }
  34. close(a.QuitChan)
  35. return nil
  36. }
  37. func (a *A) record() {
  38. //file_one, err := os.OpenFile(csvOnePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0755)
  39. file_one, err := os.Create(csvOnePath)
  40. if err != nil {
  41. fmt.Println(err)
  42. return
  43. }
  44. writer := csv.NewWriter(file_one)
  45. // writer, closeFileFunc, err := NewCsvWriter(csvOnePath)
  46. if err != nil {
  47. fmt.Println(err)
  48. return
  49. }
  50. header := []string{&quot;this&quot;, &quot;is&quot;, &quot;a&quot;, &quot;test&quot;}
  51. err = writer.Write(header)
  52. if err != nil {
  53. fmt.Println(err)
  54. return
  55. }
  56. ticker := time.NewTicker(10*time.Second)
  57. for {
  58. select {
  59. case t := &lt;-ticker.C:
  60. err = writer.Write([]string{fmt.Sprintf(&quot;%2d:%2d:%2d&quot;, t.Hour(), t.Minute(), t.Second())})
  61. if err != nil {
  62. fmt.Println(err)
  63. a.QuitChan &lt;- struct{}{}
  64. }
  65. case &lt;-a.QuitChan:
  66. ticker.Stop()
  67. writer.Flush()
  68. file_one.Close()
  69. fmt.Println(&quot;Stopped recording.&quot;)
  70. break
  71. }
  72. }
  73. }
  74. type B struct {
  75. Running 	int32 // used atomically
  76. QuitChan	chan struct{}
  77. }
  78. func NewB() *B {
  79. return &amp;B{
  80. QuitChan: make(chan struct{}),
  81. }
  82. }
  83. func (b *B) Start() error {
  84. if ok := atomic.CompareAndSwapInt32(&amp;b.Running, 0, 1); !ok {
  85. return fmt.Errorf(&quot;Cannot start service B: service already started&quot;)
  86. }
  87. go b.record()
  88. return nil
  89. }
  90. func (b *B) Stop() error {
  91. if ok := atomic.CompareAndSwapInt32(&amp;b.Running, 1, 0); !ok {
  92. return fmt.Errorf(&quot;Cannot stop service B: service already stopped&quot;)
  93. }
  94. close(b.QuitChan)
  95. return nil
  96. }
  97. func writeMsgToReport(report *csv.Writer, msg string) error {
  98. ct := time.Now()
  99. timestamp := fmt.Sprintf(&quot;%2d:%2d:%2d&quot;, ct.Hour(), ct.Minute(), ct.Second())
  100. return report.Write([]string{timestamp, msg})
  101. }
  102. func (b *B) record() {
  103. //file_two, err := os.OpenFile(csvTwoPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0666)
  104. file_two, err := os.Create(csvTwoPath)
  105. if err != nil {
  106. fmt.Println(err)
  107. return
  108. }
  109. writer := csv.NewWriter(file_two)
  110. //writer, closeFileFunc, err := NewCsvWriter(csvTwoPath)
  111. if err != nil {
  112. fmt.Println(err)
  113. return
  114. }
  115. header := []string{&quot;this&quot;, &quot;is&quot;, &quot;a&quot;, &quot;second&quot;, &quot;test&quot;}
  116. err = writer.Write(header)
  117. if err != nil {
  118. fmt.Println(err)
  119. return
  120. }
  121. ticker := time.NewTicker(1*time.Second)
  122. ticks := 0
  123. for {
  124. select {
  125. case &lt;-ticker.C:
  126. if ticks % 15 == 0 {
  127. err = writeMsgToReport(writer, &quot;YEET &quot;+strconv.Itoa(ticks))
  128. if err != nil {
  129. fmt.Println(err)
  130. b.QuitChan &lt;- struct{}{}
  131. }
  132. }
  133. ticks++
  134. case &lt;-b.QuitChan:
  135. ticker.Stop()
  136. writer.Flush()
  137. file_two.Close()
  138. fmt.Println(&quot;Stopped recording.&quot;)
  139. break
  140. }
  141. }
  142. }
  143. func main() {
  144. serviceA := NewA()
  145. err := serviceA.Start()
  146. if err != nil {
  147. fmt.Println(err)
  148. return
  149. }
  150. defer serviceA.Stop()
  151. serviceB := NewB()
  152. err = serviceB.Start()
  153. if err != nil {
  154. fmt.Println(err)
  155. return
  156. }
  157. defer serviceB.Stop()
  158. time.Sleep(600*time.Second)
  159. }

Essentially, I have two different services that run a record method in two different goroutines. They each create and write to a different csv file at different times. When I run this, the csv files are created but never have data. No errors are ever raised while running this. I read that I should use a mutex which I've implemented but this hasn't worked either. What should I do here?

答案1

得分: 1

根据评论中的详细说明，当main()函数执行完毕时，程序将退出；规范中指出：“它不会等待其他（非主）goroutine完成。”。

这意味着你的goroutine很可能不会处理关闭文件的代码，这意味着缓冲数据可能不会被写入。

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

有多种方法可以解决这个问题，但最简单的方法可能是添加一个WaitGroup，这样你的应用程序可以在终止之前等待goroutine退出。

英文:

As detailed in the comments the program will exit when main() completes; the spec states "It does not wait for other (non-main) goroutines to complete.".

This means that it is unlikely that your go routines will process the code that closes the files meaning that buffered data may not be written.

I created a simplified version of your application in the playground to demonstrate this.

There are a number of ways to fix this but the simplest is probably to add a WaitGroup so your application can wait for the go routines to exit before terminating.

huangapple
  • 本文由 发表于 2021年11月2日 02:42:22
  • 转载请务必保留本文链接：https://go.coder-hub.com/69801368.html
  • concurrency
  • csv
  • go
  • goroutine
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定