如何在 golang 中的流上发送数据时使工作并发？

今天golang学习网给大家带来了《如何在 golang 中的流上发送数据时使工作并发？》，其中涉及到的知识点包括等等，无论你是小白还是老手，都适合看一看哦~有好的建议也欢迎大家在评论留言，若是看完有所收获，也希望大家能多多点赞支持呀！一起加油学习~

我有一个 golang grpc 服务器，它具有流端点。早些时候，我按顺序完成所有工作并在流上发送，但后来我意识到我可以使工作并发，然后在流上发送。来自 grpc-go 文档：我知道我可以使工作并发，但你不能使流上的发送并发，所以我得到了下面的代码来完成这项工作。

下面是我的流端点中的代码，它以流方式将数据发送回客户端。这会同时完成所有工作。

// get "allCids" from lot of files and load in memory.
allCids := .....
var data = allCids.([]int64)
out := make(chan *custPbV1.CustomerResponse, len(data))

wg := &sync.WaitGroup{}
wg.Add(len(data))
go func() {
  wg.Wait()
  close(out)
}()

for _, cid := range data {
  go func (id int64) {
    defer wg.Done()
    pd := repo.GetCustomerData(strconv.FormatInt(cid, 10))
    if !pd.IsCorrect {
      return
    }
    resources := us.helperCom.GenerateResourceString(pd)
    val, err := us.GenerateInfo(clientId, resources, cfg)
    if err != nil {
      return
    }
    out <- val
  }(cid)
}

for val := range out {
  if err := stream.Send(val); err != nil {
    log.Printf("send error %v", err)
  }
}

现在我遇到的问题是 data 切片的大小可能约为一百万个，所以我不想生成百万个 go 例程来完成这项工作。我该如何处理这里的情况？如果我使用 len(data) 代替 100 那么这对我有用还是我也需要在 100 子数组中对数据进行切片？我只是很困惑处理这个问题的最佳方法是什么？

我最近开始使用 golang，所以如果我的上述代码在并发时有任何错误，请原谅我。

正确答案

请检查此伪代码

func main() {

    works := make(chan int, 100)
    errChan := make(chan error, 100)
    out := make(chan *custPbV1.CustomerResponse, 100)

    // spawn fixed workers
    var workerWg sync.WaitGroup
    for i := 0; i < 100; i++ {
        workerWg.Add(1)
        go worker(&workerWg, works, errChan, out)
    }

    // give input
    go func() {
        for _, cid := range data {
            // this will be blocked if all the workers are busy and no space is left in the channel.
            works <- cid
        }
        close(works)
    }()

    var analyzeResults sync.WaitGroup
    analyzeResults.Add(2)

    // process errors
    go func() {
        for err := range errChan {
            log.Printf("error %v", err)
        }
        analyzeResults.Done()
    }()

    // process outout
    go func() {
        for val := range out {
            if err := stream.Send(val); err != nil {
                log.Printf("send error %v", err)
            }
        }
        analyzeResults.Done()
    }()

    workerWg.Wait()
    close(out)
    close(errChan)
    analyzeResults.Wait()
}

func worker(job *sync.WaitGroup, works chan int, errChan chan error, out chan *custPbV1.CustomerResponse) {
    defer job.Done()
    // Idle worker takes the work from this channel. 
    for cid := range works {
        pd := repo.GetCustomerData(strconv.FormatInt(cid, 10))
        if !pd.IsCorrect {
            errChan <- errors.New(fmt.Sprintf("pd %d is incorrect", pd))
            // we can not return here as the total number of workers will be reduced. If all the workers does this then there is a chance that no workers are there to do the job
            continue
        }
        resources := us.helperCom.GenerateResourceString(pd)
        val, err := us.GenerateInfo(clientId, resources, cfg)
        if err != nil {
            errChan <- errors.New(fmt.Sprintf("got error", err))
            continue
        }
        out <- val
    }
}

说明：

这是一个工作池实现，我们在其中生成固定数量的 goroutine（此处为 100 个工作人员）来执行相同的工作（此处为 getcustomerdata() 和generateinfo()），但输入数据不同（此处为 cid）。这里的100个worker并不意味着它是并行的而是并发的（取决于gomaxprocs）。如果一个worker正在等待io结果（基本上是一些阻塞操作），那么该特定的goroutine将进行上下文切换，并且其他worker goroutine有机会执行。但是，增加 goroutuines（工作线程）可能不会提供太多性能，但可能会导致通道上的争用，因为更多的工作线程正在等待该通道上的输入作业。

相对于将 100 万个数据拆分为子切片的好处就是这样。假设我们有 1000 个工作岗位和 100 名工人。每个工人将被分配到工作 1-10、11-20 等...如果前 10 个工作比其他工作花费更多时间怎么办？在这种情况下，第一个工作人员超载，其他工作人员将完成任务并处于空闲状态，即使有待处理的任务。因此，为了避免这种情况，这是最好的解决方案，因为闲置的工人将接手下一份工作。因此，与其他工人相比，没有工人的超载程度更高

理论要掌握，实操不能落！以上关于《如何在 golang 中的流上发送数据时使工作并发？》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！