代码之家 › 专栏 › 技术社区 › MikeKlemin

Go:将许多缓慢的API查询导入单个SQL事务

goroutine channel go

MikeKlemin · 技术社区 · 8 年前

我有N个缓慢的API查询和一个数据库连接,我希望有一个缓冲通道,在那里会有响应,还有一个数据库事务,我将用它来写数据。我只能用下面的例子来描述信号量:

    func myFunc(){
      //10 concurrent API calls
      sem := make(chan bool, 10) 
     //A concurrent safe map as buffer
      var myMap  MyConcurrentMap 

      for i:=0;i<N;i++{
        sem<-true
        go func(i int){
          defer func(){<-sem}()
          resp:=slowAPICall(fmt.Sprintf("http://slow-api.me?%d",i))
          myMap.Put(resp)
        }(i)
      }

      for j=0;j<cap(sem);j++{
        sem<-true
      }
      tx,_ := db.Begin()    
      for data:=range myMap{
       tx.Exec("Insert data into database")
      }
      tx.Commit()
}

编辑: 好的,我提供了以下解决方案,这样我就不需要缓冲区映射,所以一旦数据到达 resp 通道的数据打印或可用于插入到数据库,它的工作,我仍然不确定是否一切正常,最后没有比赛。

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

//Gloab waitGroup
var wg sync.WaitGroup

func init() {
    //just for fun sake, make rand seeded
    rand.Seed(time.Now().UnixNano())
}

//Emulate a slow API call
func verySlowAPI(id int) int {
    n := rand.Intn(5)
    time.Sleep(time.Duration(n) * time.Second)
    return n
}

func main() {
    //Amount of tasks
    N := 100

    //Concurrency level
    concur := 10

    //Channel for tasks
    tasks := make(chan int, N)

    //Channel for responses
    resp := make(chan int, 10)

    //10 concurrent groutinezs
    wg.Add(concur) 
    for i := 1; i <= concur; i++ {
        go worker(tasks, resp)
    }

    //Add tasks
    for i := 0; i < N; i++ {
        tasks <- i
    }

    //Collect data from goroutiens
    for i := 0; i < N; i++ {
        fmt.Printf("%d\n", <-resp)
    }

    //close the tasks channel
    close(tasks)

    //wait till finish
    wg.Wait()

}

func worker(task chan int, resp chan<- int) {
    defer wg.Done()
    for {
        task, ok := <-task
        if !ok {
            return
        }
        n := verySlowAPI(task)
        resp <- n
    }
}

2 回复 | 直到 8 年前

Adrian 8 年前

信号量不需要使用通道, sync.WaitGroup 等待一组例程完成。

如果您使用通道限制吞吐量,最好使用工作池,并使用通道将作业传递给工作人员:

type job struct {
    i int
}

func myFunc(N int) {
    // Adjust as needed for total number of tasks
    work := make(chan job, 10)
    // res being whatever type slowAPICall returns
    results := make(chan res, 10)
    resBuff := make([]res, 0, N)

    wg := new(sync.WaitGroup)

    // 10 concurrent API calls
    for i = 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            for j := range work {
                resp := slowAPICall(fmt.Sprintf("http://slow-api.me?%d", j.i))
                results <- resp
            }
            wg.Done()
        }()
    }

    go func() {
        for r := range results {
            resBuff = append(resBuff, r)
        }
    }

    for i = 0; i < N; i++ {
        work <- job{i}
    }
    close(work)

    wg.Wait()
    close(results)
}

Pavlo Strokov 8 年前

也许这对你有用。现在,您可以摆脱并发映射。下面是一段代码片段:

func myFunc() {
    //10 concurrent API calls
    sem := make(chan bool, 10)
    respCh := make(chan YOUR_RESP_TYPE, 10)
    var responses []YOUR_RESP_TYPE

    for i := 0; i < N; i++ {
        sem <- true
        go func(i int) {
            defer func() {
                <-sem
            }()
            resp := slowAPICall(fmt.Sprintf("http://slow-api.me?%d",i))
            respCh <- resp
        }(i)
    }

    respCollected := make(chan struct{})
    go func() {
        for i := 0; i < N; i++ {
            responses = append(responses, <-respCh)
        }
        close(respCollected)
    }()

    <-respCollected
    tx,_ := db.Begin()
    for _, data := range responses {
        tx.Exec("Insert data into database")
    }
    tx.Commit()
}