当前位置： > > > > 如何在 golang 中的流上发送数据时使工作并发？
来源：stackoverflow
2024-04-22 12:27:34
0浏览
收藏
今天给大家带来了《如何在 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 中的流上发送数据时使工作并发？》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注公众号吧！
