Go实现并发扇入,批量扇出功能

uuid · · 495 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

## 背景 有上万台边缘机器,每台都会有多个agent客户端,并且每个agent都会同时向中心系统上报数据,由于上报数据频繁,并发量也大,每个agent都频繁和中心建立连接,导致中心压力非常大,所以需要对此进行优化,对每台机器上的agent上报数据做聚合,批量进行上报,减少边缘和中心的上报频率,从而减轻中心压力。 ## 方案 - 边缘增加一个batch中间件,所有agent上报的数据由原来直接向中心上报变为向batch中间件上报 - batch中间件会聚合请求,再分批向中心上报 - 上报的方式也由原来的同步也改为异步,agent只把数据提交到batch就结束 核心就是实现一个数据并发扇入,分批扇出的功能: ![image.png](https://static.studygolang.com/220306/5a978f675d2eb015d52cec75e0942fe7.png) 为了能优化分批的效果,batch中间件会有一个缓存队列,agent提交的数据会先放入到队列中,然后消费端从队列中取出数据,分批上报到中心系统,由于agent和batch中间件属于本地通讯,提交数据又是先放到队列,所以队列中的数据就容易堆积,达到分批的前提条件,有堆积数据,我就可以开始分批处理,完整逻辑如下: 1. 队列中没数据时等待 2. 有数据时,不断从队列中取出数据,当数量达到分批数量时,调用中心接口上报。 3. 数量**没达到**分批数量,但是队列中又**没有**数据时,也是**会**上报。 4. 上报结束后,又跳到2步骤,当都没数据时,就会跳到1步骤继续监听 注:因为即使数据没达到批次数量,当队列没数据时还是会上报,所以当上报数据少,且都是间隔上报,也没有并发,这时就容易出现一个批次就只有一个的情况,分批的就**没什么效果**。 接下来我们开始实现核心分批功能: ```go /** * @Title 分批处理器 * @Description 将in中的数据取出分批放入out。适合in有大量数据并发或快速写入,需要分批处理的场景 * @Author hyman * @Date 2022-03-05 **/ package batch func New(in <-chan []byte, size int) <-chan [][]byte { if size < 2 { panic("nonsense! batch size less then 2") } out := make(chan [][]byte) go func() { defer close(out) loopBatch(in, out, size) }() return out } func loopBatch(in <-chan []byte, out chan<- [][]byte, size int) { var batch [][]byte for d := range in { batch = append(batch, d) more := true // 默认认为还有数据 for more { select { default: more = false // in 没有数据 case d, ok := <-in: if ok { batch = append(batch, d) } else { more = false // in关闭,且队列里没有数据 } } l := len(batch) // more = false没数据,同时batch中也没数据,则break到外层for等待数据 if !more && l == 0 { break } // 如果more = true队列可能还有数据,当batch里的数量小于批次数量,则继续内层for尝试从in取数据 if more && l < size { continue } // 当in没数据或数量达到分批上限时,发送给out out <- batch batch = [][]byte{} // 重置 } } } ``` 核心代码不到四十行,使用一个:in <-chan []byte带缓冲的chan来接收数据,使用一个out的chan作为批量输出,只需在外层再简单包裹一层业务逻辑,就可以满足分批需求。 核心的组装批次的逻辑放在了一个协程中,巧妙的使用两个for来读取in,实现没数据时阻塞和有数据时分批上报: - 第一个for用来监听in是否有数据,当队列无数据时会pending,in被关闭时就自动结束for,协程结束。 - 第二个for巧妙用到了select的default功能,可以在不阻塞的情况下判断队列是否还有数据,实现组装分批数据。 测试 ```go func TestBatchHttp(t *testing.T) { var in = make(chan []byte, 10000) // 模拟结束请求 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if b, err := ioutil.ReadAll(r.Body); err == nil { in <- b } w.Write([]byte("success")) })) var wg sync.WaitGroup n :=time.Now() // 模拟并发上报100次 for i := 0; i < 100; i++ { wg.Add(1) _i:=i go func() { body, _ := json.Marshal(map[string]string{"say": "hello " + strconv.Itoa(_i)}) resp, err := http.Post(srv.URL, "application/json", bytes.NewReader(body)) if err != nil { t.Fatal(err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { t.Fatal(resp.Status) } }() } t.Log("post cost time = ", time.Now().Sub(n)) // 5个一批 bats := batch.New(in, 5) // 分批上报的协程数 batchWorker := 1 for i := 0; i < batchWorker; i++ { go func() { for dd := range bats { // 分批调用中心接口上报数据 for _, d := range dd { fmt.Print(string(d)) wg.Done() } fmt.Println() time.Sleep(10* time.Millisecond) // 模拟消费耗时 } }() } wg.Wait() } ``` 打印结果: ``` batch_test.go:61: post cost time = 240.333µs {"say":"hello 92"} {"say":"hello 9"} {"say":"hello 10"}{"say":"hello 79"}{"say":"hello 98"}{"say":"hello 0"}{"say":"hello 80"} {"say":"hello 99"}{"say":"hello 93"}{"say":"hello 4"}{"say":"hello 8"}{"say":"hello 5"} {"say":"hello 11"}{"say":"hello 47"}{"say":"hello 12"}{"say":"hello 7"}{"say":"hello 81"} {"say":"hello 6"}{"say":"hello 48"}{"say":"hello 13"}{"say":"hello 2"}{"say":"hello 49"} {"say":"hello 14"}{"say":"hello 75"}{"say":"hello 77"}{"say":"hello 51"}{"say":"hello 50"} {"say":"hello 74"}{"say":"hello 73"}{"say":"hello 27"}{"say":"hello 52"}{"say":"hello 76"} {"say":"hello 72"}{"say":"hello 28"}{"say":"hello 41"}{"say":"hello 53"}{"say":"hello 94"} {"say":"hello 54"}{"say":"hello 15"}{"say":"hello 95"}{"say":"hello 16"}{"say":"hello 56"} {"say":"hello 61"}{"say":"hello 58"}{"say":"hello 43"}{"say":"hello 85"}{"say":"hello 84"} {"say":"hello 89"}{"say":"hello 78"}{"say":"hello 96"}{"say":"hello 1"}{"say":"hello 82"} {"say":"hello 26"}{"say":"hello 57"}{"say":"hello 3"}{"say":"hello 46"}{"say":"hello 83"} {"say":"hello 40"}{"say":"hello 55"}{"say":"hello 29"}{"say":"hello 65"}{"say":"hello 33"} {"say":"hello 60"}{"say":"hello 42"}{"say":"hello 30"}{"say":"hello 62"}{"say":"hello 34"} {"say":"hello 64"}{"say":"hello 66"}{"say":"hello 88"}{"say":"hello 91"}{"say":"hello 35"} {"say":"hello 38"}{"say":"hello 39"}{"say":"hello 87"}{"say":"hello 18"}{"say":"hello 59"} {"say":"hello 68"}{"say":"hello 19"}{"say":"hello 63"}{"say":"hello 20"}{"say":"hello 21"} {"say":"hello 86"}{"say":"hello 17"}{"say":"hello 24"}{"say":"hello 70"}{"say":"hello 69"} {"say":"hello 90"}{"say":"hello 22"}{"say":"hello 71"}{"say":"hello 25"}{"say":"hello 44"} {"say":"hello 45"}{"say":"hello 36"}{"say":"hello 23"}{"say":"hello 37"}{"say":"hello 32"} {"say":"hello 31"}{"say":"hello 67"}{"say":"hello 97"} --- PASS: TestBatchHttp (0.26s) PASS ``` 100个请求数据都有正常打印,并且大部分数量都符合预期,但是前两行只有1个,并没有到达分批数量,这是为什么?! 原因是刚开始分批速度大于提交速度,取出1个数据时队列就没有数据,所以第一批也就只有1个数据,第二批也是一样,由于批次消费就只有开一个协程,且每次消费都time.Sleep(10* time.Millisecond),消费过慢,于是数据逐渐堆积,满足批次数量,后面打印就正常。 PS:要验证这一说法,可以在”bats := batch.New(in, 5)“前面增加”time.Sleep(10* time.Millisecond)“让队列里数据有足够数据,这时打印就会符合预期。 这里的batchWorker只有1个,也就是说是向中心是串行上报,这个可以根据实际需求修改,通过开启多个协程来提高上报速度。 但是,有时候反而不满足需求,这样做会并发上报数据,增加中心压力,同时由于消费过快,反而使队列中的数量就不容易达到分批数量,分批效果就减少,只是加快队列消费而已。 ## 总结 Go的chan功能非常强大,各种并发处理都能看到他的身影,也是Go推荐的方式: > Do not communicate by sharing memory; instead, share memory by communicating. 灵活借助chan并发安全,既能阻塞又能select default非阻塞的特性,可以优雅实现很多并发控制的需求 我的博客:[Go实现并发扇入,批量扇出功能 | 艺术码农的小栈](https://itart.cn/blogs/2022/practice/go-chan-aggregate-batch.html)

有疑问加站长微信联系(非本文作者))

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

495 次点击  ∙  1 赞  
加入收藏 微博
1 回复  |  直到 2022-03-07 09:44:55
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传