背景
有上万台边缘机器，每台都会有多个 agent 客户端，并且每个 agent 都会同时向中心系统上报数据，由于上报数据频繁，并发量也大，每个 agent 都频繁和中心建立连接，导致中心压力非常大，所以需要对此进行优化，对每台机器上的 agent 上报数据做聚合，批量进行上报，减少边缘和中心的上报频率，从而减轻中心压力。
方案
核心就是实现一个数据并发扇入，分批扇出的功能：
为了能优化分批的效果，batch 中间件会有一个缓存队列，agent 提交的数据会先放入到队列中，然后消费端从队列中取出数据，分批上报到中心系统，由于 agent 和 batch 中间件属于本地通讯，提交数据又是先放到队列，所以队列中的数据就容易堆积，达到分批的前提条件，有堆积数据，我就可以开始分批处理，完整逻辑如下：
队列中没数据时等待
有数据时，不断从队列中取出数据，当数量达到分批数量时，调用中心接口上报。
数量没达到分批数量，但是队列中又没有数据时，也是会上报。
上报结束后，又跳到 2 步骤，当都没数据时，就会跳到 1 步骤继续监听
注：因为即使数据没达到批次数量，当队列没数据时还是会上报，所以当上报数据少，且都是间隔上报，也没有并发，这时就容易出现一个批次就只有一个的情况，分批的就没什么效果。
接下来我们开始实现核心分批功能：
/** * @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，实现没数据时阻塞和有数据时分批上报：
测试
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 非阻塞的特性，可以优雅实现很多并发控制的需求
评论