使用多个接收器来接收一个频道的内容

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

Using multiple receivers for a channel

问题

我正在尝试使用go-json-rest在golang中编写一个REST服务。

该服务的目的只是将接收到的数据转换为CSV格式并记录下来。
由于负载可能很重，我希望使用goroutines进行日志记录。
目前我已经创建了四个LogWorker（goroutine）
每个goroutine将CSV记录到单独的文件中。

当我执行代码时，日志总是从最后一个goroutine触发。我在我的日志文件夹中只看到一个文件，来自第四个routine。

这是我的服务器代码：

  1. package main
  3. import (
  4.     "github.com/ant0ine/go-json-rest/rest"
  5.     "log"
  6.     "net/http"
  7.     "strconv"
  8.     "time"
  9. )
  11. const workerCount = 4
  12. var evChannel = make(chan Event)
  13. var workers = make([]*LogWorker, workerCount)
  14. const maxLogFileSize = 100 // In MB
  15. const maxLogFileBackups = 30
  16. const maxLogFileAge = 5
  17. const logFileName = "/home/sam/tmp/go_logs/event_"
  19. func main() {
  20.     // 初始化 workers
  21.     // 创建了四个 workers
  22.     for i := 0; i < workerCount; i++ {
  23.         var fileName = logFileName + strconv.Itoa(i)
  24.         workers[i] = NewLogWorker(fileName, maxLogFileSize, maxLogFileBackups, maxLogFileAge)
  25.         go workers[i].Work(evChannel)
  26.     }    
  28.     // 初始化 REST API
  29.     api := rest.NewApi()
  30.     //api.Use(rest.DefaultDevStack...)
  31.     api.Use(rest.DefaultCommonStack...)
  32.     router, err := rest.MakeRouter(
  33.         rest.Post("/events", StoreEvents),
  34.     )
  35.     if err != nil {
  36.         log.Fatal(err)
  37.     }
  38.     api.SetApp(router)
  39.     log.Fatal(http.ListenAndServe(":4545", api.MakeHandler()))
  40. }
  42. func StoreEvents(w rest.ResponseWriter, r *rest.Request) {
  43.     event := Event{}
  44.     err := r.DecodeJsonPayload(&event)
  45.     if err != nil {
  46.         rest.Error(w, err.Error(), http.StatusInternalServerError)
  47.         return
  48.     }
  49.     // TODO : Add validation if needed
  50.     // Add code to parse the request and add further information to event 
  51.     // log.Println()
  52.     select {
  53.         case evChannel <- event:
  54.         case <- time.After(5 * time.Second):
  55.       // throw away the message, so sad
  56.     }    
  57.     // evChannel <- event
  58.     //log.Println(Csv(event))
  59.     w.WriteHeader(http.StatusOK)
  60. }

这是我的worker代码：

  1. package main
  3. import (
  4.     "gopkg.in/natefinch/lumberjack.v2"
  5.     "log"
  6.     "fmt"
  7. )
  9. type LogWorker struct {
  10.     FileName string
  11.     MaxSize int // In megabytes
  12.     MaxBackups int // No of backups per worker
  13.     MaxAge int // maximum number of days to retain old log files
  14. }
  16. func NewLogWorker(fileName string, maxSize int, maxBackups int, maxAge int) (lw *LogWorker) {
  17.     return &LogWorker {fileName, maxSize, maxBackups, maxAge}
  18. }
  20. func (lw *LogWorker) Work(evChannel chan Event) {
  21.     fmt.Println(lw.FileName)
  22.     log.SetOutput(&lumberjack.Logger {
  23.         Filename:   lw.FileName,
  24.         MaxSize:    lw.MaxSize,
  25.         MaxBackups: lw.MaxBackups,
  26.         MaxAge:     lw.MaxAge,
  27.     })
  28.     log.SetFlags(0)
  30.     for {
  31.         event := <- evChannel
  32.         log.Println(Csv(event))
  33.     }
  34. }

请注意，event是一个包含一些字符串字段的结构体。
在SO上已经有一个类似的问题。当我尝试在playground中执行goroutine时，它仍然打印出最后一个goroutine的值。提供的答案中有一些wait.Done。由于我的worker需要持续运行，我认为我不能使用它。

请帮助我找出为什么我的所有goroutines（LogWorkers）都没有被使用？

英文:

I am trying write a REST service in golang using go-json-rest

The purpose of the service is just to convert the received data to CSV and log it.
Since the load may be heavy, I would like to do the logging using goroutines.
Currently I have created four LogWorkers(goroutine)
Each goroutine will log the CSV into separate files.

When I execute the code, the log is always triggered from the last goroutine. I see a single file created in my log folder which is from fourth routine.

Here is my server code

  1. package main
  2. import (
  3. &quot;github.com/ant0ine/go-json-rest/rest&quot;
  4. &quot;log&quot;
  5. &quot;net/http&quot;
  6. &quot;strconv&quot;
  7. &quot;time&quot;
  8. )
  9. const workerCount = 4
  10. var evChannel = make(chan Event)
  11. var workers = make([]*LogWorker, workerCount)
  12. const maxLogFileSize = 100 // In MB
  13. const maxLogFileBackups = 30
  14. const maxLogFileAge = 5
  15. const logFileName = &quot;/home/sam/tmp/go_logs/event_&quot;
  16. func main() {
  17. // Initialize workers
  18. // Four workers is being created
  19. for i := 0; i &lt; workerCount; i++ {
  20. var fileName = logFileName + strconv.Itoa(i)
  21. workers[i] = NewLogWorker(fileName, maxLogFileSize, maxLogFileBackups, maxLogFileAge)
  22. go workers[i].Work(evChannel)
  23. }    
  24. // Initialize REST API
  25. api := rest.NewApi()
  26. //api.Use(rest.DefaultDevStack...)
  27. api.Use(rest.DefaultCommonStack...)
  28. router, err := rest.MakeRouter(
  29. rest.Post(&quot;/events&quot;, StoreEvents),
  30. )
  31. if err != nil {
  32. log.Fatal(err)
  33. }
  34. api.SetApp(router)
  35. log.Fatal(http.ListenAndServe(&quot;:4545&quot;, api.MakeHandler()))
  36. }
  37. func StoreEvents(w rest.ResponseWriter, r *rest.Request) {
  38. event := Event{}
  39. err := r.DecodeJsonPayload(&amp;event)
  40. if err != nil {
  41. rest.Error(w, err.Error(), http.StatusInternalServerError)
  42. return
  43. }
  44. // TODO : Add validation if needed
  45. // Add code to parse the request and add further information to event 
  46. // log.Println()
  47. select {
  48. case evChannel &lt;- event:
  49. case &lt;- time.After(5 * time.Second):
  50. // throw away the message, so sad
  51. }    
  52. // evChannel &lt;- event
  53. //log.Println(Csv(event))
  54. w.WriteHeader(http.StatusOK)
  55. }

here is my worker code

  1. package main
  2. import (
  3. &quot;gopkg.in/natefinch/lumberjack.v2&quot;
  4. &quot;log&quot;
  5. &quot;fmt&quot;
  6. )
  7. type LogWorker struct {
  8. FileName string
  9. MaxSize int // In megabytes
  10. MaxBackups int // No of backups per worker
  11. MaxAge int // maximum number of days to retain old log files
  12. }
  13. func NewLogWorker(fileName string, maxSize int, maxBackups int, maxAge int) (lw *LogWorker) {
  14. return &amp;LogWorker {fileName, maxSize, maxBackups, maxAge}
  15. }
  16. func (lw *LogWorker) Work(evChannel chan Event) {
  17. fmt.Println(lw.FileName)
  18. log.SetOutput(&amp;lumberjack.Logger {
  19. Filename:   lw.FileName,
  20. MaxSize:    lw.MaxSize,
  21. MaxBackups: lw.MaxBackups,
  22. MaxAge:     lw.MaxAge,
  23. })
  24. log.SetFlags(0)
  25. for {
  26. event := &lt;- evChannel
  27. log.Println(Csv(event))
  28. }
  29. }

Please note that event is a struct which contains some string fields.
Already there is a similar question in SO. When I tried to execute the goroutine in playground, it still prints the value from last go routine. The answer provided has some wait.Done. As my worker needs to run continuously I don't I think I can use it.

Please help me to find why my all goroutines (LogWorkers) are not used?

答案1

得分: 4

你正在为每个goroutine设置日志包的默认全局记录器的输出。

你可能想要做更多的事情，像这样：

  1. func (lw *LogWorker) Work(evChannel chan Event) {
  2.     fmt.Println(lw.FileName)
  3.     lg := log.New(&lumberjack.Logger {
  4.         Filename:   lw.FileName,
  5.         MaxSize:    lw.MaxSize,
  6.         MaxBackups: lw.MaxBackups,
  7.         MaxAge:     lw.MaxAge,
  8.     }, "", 0)
  10.     for {
  11.         event := <- evChannel
  12.         lg.Println(Csv(event))
  13.     }
  14. }

这样将为每个goroutine提供一个记录器。

在你的版本中，你可能只有最后一个执行的记录器（可能是最后一个生成的goroutine，但不能保证）。

为了进一步改进，你可能还想将你的for循环写成：

  1. for event := range evChannel {
  2.     lg.Println(Csv(event))
  3. }

这样，当通道关闭时，它将终止goroutine，而不是在关闭的通道中旋转空值。参考这里

英文:

You are setting the log package's default global logger's output in each goroutine.

You probably want to do something more like:

  1. func (lw *LogWorker) Work(evChannel chan Event) {
  2. fmt.Println(lw.FileName)
  3. lg := log.New(&amp;lumberjack.Logger {
  4. Filename:   lw.FileName,
  5. MaxSize:    lw.MaxSize,
  6. MaxBackups: lw.MaxBackups,
  7. MaxAge:     lw.MaxAge,
  8. }, &quot;&quot;, 0)
  9. for {
  10. event := &lt;- evChannel
  11. lg.Println(Csv(event))
  12. }
  13. }

this will give you a logger per goroutine.

In your version, you likely just had the last one to execute (probably last goroutine spawned, but not guaranteed)

And to improve a bit more, you probably also want your for loop written as:

  1. for event := range evChannel {
  2. lg.Println(Csv(event))
  3. }

This way, it will terminate the goroutine when the channel is closed rather than spin on empty values coming out of a closed channel. See here for reference

huangapple
  • 本文由 发表于 2015年11月28日 13:40:02
  • 转载请务必保留本文链接：https://go.coder-hub.com/33968098.html
  • channel
  • concurrency
  • csv
  • go
匿名

发表评论

匿名网友

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

确定