dthp96899
2015-09-02 13:02
浏览 68
已采纳

Golang:限制阻塞操作的并发级别

I have the following scenario: I am receiving a message on a channel telling me to upload a file. The upload is made by the blocking function uploadToServer. The zipGen channel may receive several messages per second, and I want to upload maximum 5 files simultaneously (not more, but possibly less - depending on how many messages are sent on zipGen by a third worker that is out of the scope of this question).

The listenToZips function runs inside a go routine (go listenToZips() on the file's init function):

func listenToZips() {
    for {
        select {
        case zip := <-zipGen:
          uploadToServer(zip) // this is blocking
        }
    }
}

If I launch go uploadToServer(zip) instead of just uploadToServer(zip) - I get too much concurrency (so for example my program will try to upload 10 files at the same time, but I want a maximum of 5).

On the other hand, without go uploadToServer(zip) (just using uploadToServer(zip) like in the above function), I only upload one file at a time (since the uploadToServer(zip) is blocking).

How can I achieve this level of control to allow me a max upload of 5 files simultaneously?

Thanks!

  • 写回答
  • 好问题 提建议
  • 关注问题
  • 收藏
  • 邀请回答

2条回答 默认 最新

  • dougou8458 2015-09-02 13:07
    已采纳

    The simplest option - prespawn N goroutines that take input from the channel, and upload it, in a loop. In each goroutine's context the operation will be blocking, but N goroutines do this. Only one goroutine will receive each message, of course.

    func listenToZips(concurrent int) {
    
        for i:=0; i < concurrent; i++ {
    
          // spawn a listener goroutine
          go func() {
    
             for {
                select {
                case zip := <-zipGen:
                   uploadToServer(zip) // this is blocking
                }
             }
          }()
    
       }
    
    }
    

    Of course you can then add stop condition, probably using a different channel, but the basic idea is just the same.

    已采纳该答案
    评论
    解决 无用
    打赏 举报
  • duanlu6268 2017-05-13 18:35

    try this: https://github.com/korovkin/limiter

     limiter := NewConcurrencyLimiter(10)
     limiter.Execute(func() {
            uploadToServer()
     })
     limiter.Wait()
    
    评论
    解决 无用
    打赏 举报

相关推荐 更多相似问题