csv - 为一个 channel 使用多个接收器

Question

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

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

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

这是我的服务器代码

package main

import (
    "github.com/ant0ine/go-json-rest/rest"
    "log"
    "net/http"
    "strconv"
    "time"
)

const workerCount = 4
var evChannel = make(chan Event)
var workers = make([]*LogWorker, workerCount)
const maxLogFileSize = 100 // In MB
const maxLogFileBackups = 30
const maxLogFileAge = 5
const logFileName = "/home/sam/tmp/go_logs/event_"

func main() {
    // Initialize workers
    // Four workers is being created
    for i := 0; i < workerCount; i++ {
        var fileName = logFileName + strconv.Itoa(i)
        workers[i] = NewLogWorker(fileName, maxLogFileSize, maxLogFileBackups, maxLogFileAge)
        go workers[i].Work(evChannel)
    }    

    // Initialize REST API
    api := rest.NewApi()
    //api.Use(rest.DefaultDevStack...)
    api.Use(rest.DefaultCommonStack...)
    router, err := rest.MakeRouter(
        rest.Post("/events", StoreEvents),
    )
    if err != nil {
        log.Fatal(err)
    }
    api.SetApp(router)
    log.Fatal(http.ListenAndServe(":4545", api.MakeHandler()))
}

func StoreEvents(w rest.ResponseWriter, r *rest.Request) {
    event := Event{}
    err := r.DecodeJsonPayload(&event)
    if err != nil {
        rest.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    // TODO : Add validation if needed
    // Add code to parse the request and add further information to event 
    // log.Println()
    select {
        case evChannel <- event:
        case <- time.After(5 * time.Second):
      // throw away the message, so sad
    }    
    // evChannel <- event
    //log.Println(Csv(event))
    w.WriteHeader(http.StatusOK)
}

这是我的 worker 代码

package main

import (
    "gopkg.in/natefinch/lumberjack.v2"
    "log"
    "fmt"
)

type LogWorker struct {
    FileName string
    MaxSize int // In megabytes
    MaxBackups int // No of backups per worker
    MaxAge int // maximum number of days to retain old log files
}

func NewLogWorker(fileName string, maxSize int, maxBackups int, maxAge int) (lw *LogWorker) {
    return &LogWorker {fileName, maxSize, maxBackups, maxAge}
}

func (lw *LogWorker) Work(evChannel chan Event) {
    fmt.Println(lw.FileName)
    log.SetOutput(&lumberjack.Logger {
        Filename:   lw.FileName,
        MaxSize:    lw.MaxSize,
        MaxBackups: lw.MaxBackups,
        MaxAge:     lw.MaxAge,
    })
    log.SetFlags(0)

    for {
        event := <- evChannel
        log.Println(Csv(event))
    }
}

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

请帮我找出为什么我的所有 goroutines (LogWorkers) 都没有被使用？

Answer 1

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

你可能想做一些更像是:

func (lw *LogWorker) Work(evChannel chan Event) {
    fmt.Println(lw.FileName)
    lg := log.New(&lumberjack.Logger {
        Filename:   lw.FileName,
        MaxSize:    lw.MaxSize,
        MaxBackups: lw.MaxBackups,
        MaxAge:     lw.MaxAge,
    }, "", 0)

    for {
        event := <- evChannel
        lg.Println(Csv(event))
    }
}

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

在您的版本中，您可能只有最后一个要执行(可能是最后生成的 goroutine，但不能保证)

为了进一步改进，您可能还希望将 for 循环编写为:

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

这样，它将在 channel 关闭时终止 goroutine，而不是对来自已关闭 channel 的空值进行自旋。 See here供引用