英文:
Concurrent file parsing and inserting into Elastic Search
问题
我最近在使用Go语言编写代码，并编写了一个小脚本，用于解析日志文件并将其插入到Elasticsearch中。对于每个文件，我都会启动一个goroutine，代码如下：
var wg sync.WaitGroup
wg.Add(len(files))
for _, file := range files {
go func(f os.FileInfo){
defer wg.Done()
ProcessFile(f.Name(), config.OriginFilePath, config.WorkingFilePath, config.ArchiveFilePath,fmt.Sprintf("http://%v:%v", config.ElasticSearch.Host, config.ElasticSearch.Port),config.ProviderIndex, config.NetworkData)
}(file)
}
wg.Wait()
在ProcessFile函数中，我有一个向Elasticsearch发送请求的函数：
func BulkInsert(lines []string, ES *elastic.Client) (*elastic.Response, error){
r, err := ES.PerformRequest("POST", "/_bulk", url.Values{}, strings.Join(lines, "\n")+"\n")
if err != nil {
return nil, err
}
return r, nil
}
问题是，我不太理解goroutine的工作原理。我理解的是，向Elasticsearch发送请求会阻塞我的某个goroutine的执行。我尝试使用相同的方法为Elasticsearch的批量插入操作启动另一个goroutine：
WaitGroup
go func(){defer wg.Done(); BulkInsert(elems, ES);}()
wg.Wait()
然后在函数返回之前使用wg.Wait()。然而，我发现最终并不是所有的事件都被插入到Elasticsearch中。我认为这是因为goroutine在没有发送/等待批量请求完成的情况下就返回了。
我的问题是，我的解决方法是否正确？我能否获得更好的性能？
英文:
I was playing with Go recently and come up with small script which parses log files and inserts them to elastic search. For each file I spawned a goroutine like this:
var wg := sync.WaitGroup{}
wg.Add(len(files))
for _, file := range files {
go func(f os.FileInfo){
defer wg.Done()
ProcessFile(f.Name(), config.OriginFilePath, config.WorkingFilePath, config.ArchiveFilePath,fmt.Sprintf("http://%v:%v", config.ElasticSearch.Host, config.ElasticSearch.Port),config.ProviderIndex, config.NetworkData)
}(file)
}
wg.Wait()
Inside of my processFile I have function which sends to elastic search:
func BulkInsert(lines []string, ES *elastic.Client) (*elastic.Response, error){
r, err := ES.PerformRequest("POST", "/_bulk", url.Values{}, strings.Join(lines, "\n")+"\n")
if err != nil {
return nil, err
}
return r, nil
}
The problem is that I don't fully understand how goroutines work. My understanding is that sending to elastic search is blocking one of my goroutines from executing. I tried spawning another goroutine for elastic search with bulk insert with same approach:
WaitGroup, go func(){defer wg.Done(); BulkInsert(elems, ES);}()
and wg.Wait() before my function return. However, I've discovered that in the end that not all my events end up in elastic search. I think this is due to goroutines returning without ever sending/waiting for bulk request to finish.
My question is, is my approach to this problem is correct? Can I achieve better performance?
答案1
得分: 1
我可以帮你翻译这段代码。以下是翻译的内容：
> 我能获得更好的性能吗？
不清楚，这取决于接收者和发送者的能力。
> 我的问题是，我对这个问题的方法正确吗？
这段代码主要是关于Go语言中的goroutine的使用。它创建了多个goroutine来并发地发送HTTP请求。代码中有三个案例，分别展示了不同的goroutine使用方式。
第一个案例创建了10个goroutine，并发地发送10个HTTP请求。
第二个案例创建了3个goroutine，并发地发送3个HTTP请求。这里使用了一个带缓冲的通道来限制并发的数量。
第三个案例对第二个案例进行了重写，使用了一个自定义的函数parallel来简化并发的操作。
你可以根据自己的需求选择适合的方法来提高性能。
英文:
> Can I achieve better performance?
unclear, it depends of the receiver and the sender capabilities.
> My question is, is my approach to this problem is correct?
might this help you better understand go routines,
package main
import (
"fmt"
"log"
"net/http"
"sync"
"time"
)
func main() {
addr := "127.0.0.1:2074"
srv := http.Server{
Addr: addr,
Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
log.Println("hit ", r.URL.String())
<-time.After(time.Second)
log.Println("done ", r.URL.String())
}),
}
fail(unblock(srv.ListenAndServe))
jobs := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
// case 1
// it creates 10 goroutines,
// that triggers 10 // concurrent get queries
{
wg := sync.WaitGroup{}
wg.Add(len(jobs))
log.Printf("starting %v jobs\n", len(jobs))
for _, job := range jobs {
go func(job int) {
defer wg.Done()
http.Get(fmt.Sprintf("http://%v/job/%v", addr, job))
}(job)
}
wg.Wait()
log.Printf("done %v jobs\n", len(jobs))
}
log.Println()
log.Println("=================")
log.Println()
// case 2
// it creates 3 goroutines,
// that triggers 3 // concurrent get queries
{
wg := sync.WaitGroup{}
wg.Add(len(jobs))
in := make(chan string)
limit := make(chan bool, 3)
log.Printf("starting %v jobs\n", len(jobs))
go func() {
for url := range in {
limit <- true
go func(url string) {
defer wg.Done()
http.Get(url)
<-limit
}(url)
}
}()
for _, job := range jobs {
in <- fmt.Sprintf("http://%v/job/%v", addr, job)
}
wg.Wait()
log.Printf("done %v jobs\n", len(jobs))
}
log.Println()
log.Println("=================")
log.Println()
// case 2: rewrite
// it creates 6 goroutines,
// that triggers 6 // concurrent get queries
{
wait, add := parallel(6)
log.Printf("starting %v jobs\n", len(jobs))
for _, job := range jobs {
url := fmt.Sprintf("http://%v/job/%v", addr, job)
add(func() {
http.Get(url)
})
}
wait()
log.Printf("done %v jobs\n", len(jobs))
}
}
func parallel(c int) (func(), func(block func())) {
wg := sync.WaitGroup{}
in := make(chan func())
limit := make(chan bool, c)
go func() {
for block := range in {
limit <- true
go func(block func()) {
defer wg.Done()
block()
<-limit
}(block)
}
}()
return wg.Wait, func(block func()) {
wg.Add(1)
in <- block
}
}
func unblock(block func() error) error {
w := make(chan error)
go func() { w <- block() }()
select {
case err := <-w:
return err
case <-time.After(time.Millisecond):
}
return nil
}
func fail(err error) {
if err != nil {
panic(err)
}
}
outputs
$ go run main.go
2017/09/14 01:30:50 starting 10 jobs
2017/09/14 01:30:50 hit /job/0
2017/09/14 01:30:50 hit /job/4
2017/09/14 01:30:50 hit /job/5
2017/09/14 01:30:50 hit /job/2
2017/09/14 01:30:50 hit /job/9
2017/09/14 01:30:50 hit /job/1
2017/09/14 01:30:50 hit /job/3
2017/09/14 01:30:50 hit /job/7
2017/09/14 01:30:50 hit /job/8
2017/09/14 01:30:50 hit /job/6
2017/09/14 01:30:51 done /job/5
2017/09/14 01:30:51 done /job/4
2017/09/14 01:30:51 done /job/2
2017/09/14 01:30:51 done /job/0
2017/09/14 01:30:51 done /job/6
2017/09/14 01:30:51 done /job/9
2017/09/14 01:30:51 done /job/1
2017/09/14 01:30:51 done /job/3
2017/09/14 01:30:51 done /job/7
2017/09/14 01:30:51 done /job/8
2017/09/14 01:30:51 done 10 jobs
2017/09/14 01:30:51
2017/09/14 01:30:51 =================
2017/09/14 01:30:51
2017/09/14 01:30:51 starting 10 jobs
2017/09/14 01:30:51 hit /job/0
2017/09/14 01:30:51 hit /job/2
2017/09/14 01:30:51 hit /job/1
2017/09/14 01:30:52 done /job/2
2017/09/14 01:30:52 done /job/0
2017/09/14 01:30:52 done /job/1
2017/09/14 01:30:52 hit /job/3
2017/09/14 01:30:52 hit /job/4
2017/09/14 01:30:52 hit /job/5
2017/09/14 01:30:53 done /job/3
2017/09/14 01:30:53 done /job/4
2017/09/14 01:30:53 done /job/5
2017/09/14 01:30:53 hit /job/6
2017/09/14 01:30:53 hit /job/7
2017/09/14 01:30:53 hit /job/8
2017/09/14 01:30:54 done /job/6
2017/09/14 01:30:54 done /job/7
2017/09/14 01:30:54 done /job/8
2017/09/14 01:30:54 hit /job/9
2017/09/14 01:30:55 done /job/9
2017/09/14 01:30:55 done 10 jobs
2017/09/14 01:30:55
2017/09/14 01:30:55 =================
2017/09/14 01:30:55
2017/09/14 01:30:55 starting 10 jobs
2017/09/14 01:30:55 hit /job/0
2017/09/14 01:30:55 hit /job/1
2017/09/14 01:30:55 hit /job/4
2017/09/14 01:30:55 hit /job/2
2017/09/14 01:30:55 hit /job/3
2017/09/14 01:30:55 hit /job/5
2017/09/14 01:30:56 done /job/0
2017/09/14 01:30:56 hit /job/6
2017/09/14 01:30:56 done /job/1
2017/09/14 01:30:56 done /job/2
2017/09/14 01:30:56 done /job/4
2017/09/14 01:30:56 hit /job/7
2017/09/14 01:30:56 done /job/3
2017/09/14 01:30:56 hit /job/9
2017/09/14 01:30:56 hit /job/8
2017/09/14 01:30:56 done /job/5
2017/09/14 01:30:57 done /job/6
2017/09/14 01:30:57 done /job/7
2017/09/14 01:30:57 done /job/9
2017/09/14 01:30:57 done /job/8
2017/09/14 01:30:57 done 10 jobs
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论