背景

有上萬(wàn)臺(tái)邊緣機(jī)器，每臺(tái)都會(huì)有多個(gè)agent客戶端，并且每個(gè)agent都會(huì)同時(shí)向中心系統(tǒng)上報(bào)數(shù)據(jù)，由于上報(bào)數(shù)據(jù)頻繁，并發(fā)量也大，每個(gè)agent都頻繁和中心建立連接，導(dǎo)致中心壓力非常大，所以需要對(duì)此進(jìn)行優(yōu)化，對(duì)每臺(tái)機(jī)器上的agent上報(bào)數(shù)據(jù)做聚合，批量進(jìn)行上報(bào)，減少邊緣和中心的上報(bào)頻率，從而減輕中心壓力。

方案

• 邊緣增加一個(gè)batch中間件，所有agent上報(bào)的數(shù)據(jù)由原來(lái)直接向中心上報(bào)變?yōu)橄騜atch中間件上報(bào)
• batch中間件會(huì)聚合請(qǐng)求，再分批向中心上報(bào)
• 上報(bào)的方式也由原來(lái)的同步也改為異步，agent只把數(shù)據(jù)提交到batch就結(jié)束

核心就是實(shí)現(xiàn)一個(gè)數(shù)據(jù)并發(fā)扇入，分批扇出的功能：

image.png

為了能優(yōu)化分批的效果，batch中間件會(huì)有一個(gè)緩存隊(duì)列，agent提交的數(shù)據(jù)會(huì)先放入到隊(duì)列中，然后消費(fèi)端從隊(duì)列中取出數(shù)據(jù)，分批上報(bào)到中心系統(tǒng)，由于agent和batch中間件屬于本地通訊，提交數(shù)據(jù)又是先放到隊(duì)列，所以隊(duì)列中的數(shù)據(jù)就容易堆積，達(dá)到分批的前提條件，有堆積數(shù)據(jù)，我就可以開(kāi)始分批處理，完整邏輯如下：

1. 隊(duì)列中沒(méi)數(shù)據(jù)時(shí)等待
2. 有數(shù)據(jù)時(shí)，不斷從隊(duì)列中取出數(shù)據(jù)，當(dāng)數(shù)量達(dá)到分批數(shù)量時(shí)，調(diào)用中心接口上報(bào)。
3. 數(shù)量沒(méi)達(dá)到分批數(shù)量，但是隊(duì)列中又沒(méi)有數(shù)據(jù)時(shí)，也是會(huì)上報(bào)。
4. 上報(bào)結(jié)束后，又跳到2步驟，當(dāng)都沒(méi)數(shù)據(jù)時(shí)，就會(huì)跳到1步驟繼續(xù)監(jiān)聽(tīng)

注：因?yàn)榧词箶?shù)據(jù)沒(méi)達(dá)到批次數(shù)量，當(dāng)隊(duì)列沒(méi)數(shù)據(jù)時(shí)還是會(huì)上報(bào)，所以當(dāng)上報(bào)數(shù)據(jù)少，且都是間隔上報(bào)，也沒(méi)有并發(fā)，這時(shí)就容易出現(xiàn)一個(gè)批次就只有一個(gè)的情況，分批的就沒(méi)什么效果。

接下來(lái)我們開(kāi)始實(shí)現(xiàn)核心分批功能：

/**
 * @Title 分批處理器
 * @Description 將in中的數(shù)據(jù)取出分批放入out。適合in有大量數(shù)據(jù)并發(fā)或快速寫(xiě)入，需要分批處理的場(chǎng)景
 * @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 // 默認(rèn)認(rèn)為還有數(shù)據(jù)
        for more {
            select {
            default:
                more = false // in 沒(méi)有數(shù)據(jù)
            case d, ok := <-in:
                if ok {
                    batch = append(batch, d)
                } else {
                    more = false // in關(guān)閉，且隊(duì)列里沒(méi)有數(shù)據(jù)
                }
            }
            l := len(batch)
            // more = false沒(méi)數(shù)據(jù)，同時(shí)batch中也沒(méi)數(shù)據(jù)，則break到外層for等待數(shù)據(jù)
            if !more && l == 0 {
                break
            }
            // 如果more = true隊(duì)列可能還有數(shù)據(jù)，當(dāng)batch里的數(shù)量小于批次數(shù)量，則繼續(xù)內(nèi)層for嘗試從in取數(shù)據(jù)
            if more && l < size {
                continue
            }
            // 當(dāng)in沒(méi)數(shù)據(jù)或數(shù)量達(dá)到分批上限時(shí)，發(fā)送給out
            out <- batch
            batch = [][]byte{} // 重置
        }
    }
}

核心代碼不到四十行，使用一個(gè)：in <-chan []byte帶緩沖的chan來(lái)接收數(shù)據(jù)，使用一個(gè)out的chan作為批量輸出，只需在外層再簡(jiǎn)單包裹一層業(yè)務(wù)邏輯，就可以滿足分批需求。

核心的組裝批次的邏輯放在了一個(gè)協(xié)程中，巧妙的使用兩個(gè)for來(lái)讀取in，實(shí)現(xiàn)沒(méi)數(shù)據(jù)時(shí)阻塞和有數(shù)據(jù)時(shí)分批上報(bào)：

• 第一個(gè)for用來(lái)監(jiān)聽(tīng)in是否有數(shù)據(jù)，當(dāng)隊(duì)列無(wú)數(shù)據(jù)時(shí)會(huì)pending，in被關(guān)閉時(shí)就自動(dòng)結(jié)束for，協(xié)程結(jié)束。
• 第二個(gè)for巧妙用到了select的default功能，可以在不阻塞的情況下判斷隊(duì)列是否還有數(shù)據(jù)，實(shí)現(xiàn)組裝分批數(shù)據(jù)。

測(cè)試

func TestBatchHttp(t *testing.T) {
    var in = make(chan []byte, 10000)
    // 模擬結(jié)束請(qǐng)求
    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()
    // 模擬并發(fā)上報(bào)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個(gè)一批
    bats := batch.New(in, 5)
    // 分批上報(bào)的協(xié)程數(shù)
    batchWorker := 1
    for i := 0; i < batchWorker; i++ {
        go func() {
            for dd := range bats {
                // 分批調(diào)用中心接口上報(bào)數(shù)據(jù)
                for _, d := range dd {
                    fmt.Print(string(d))
                    wg.Done()
                }
                fmt.Println()
                time.Sleep(10* time.Millisecond) // 模擬消費(fèi)耗時(shí)

            }
        }()
    }
    wg.Wait()
}

打印結(jié)果：

 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個(gè)請(qǐng)求數(shù)據(jù)都有正常打印，并且大部分?jǐn)?shù)量都符合預(yù)期，但是前兩行只有1個(gè)，并沒(méi)有到達(dá)分批數(shù)量，這是為什么？！

原因是剛開(kāi)始分批速度大于提交速度，取出1個(gè)數(shù)據(jù)時(shí)隊(duì)列就沒(méi)有數(shù)據(jù)，所以第一批也就只有1個(gè)數(shù)據(jù)，第二批也是一樣，由于批次消費(fèi)就只有開(kāi)一個(gè)協(xié)程，且每次消費(fèi)都time.Sleep(10* time.Millisecond)，消費(fèi)過(guò)慢，于是數(shù)據(jù)逐漸堆積，滿足批次數(shù)量，后面打印就正常。

PS：要驗(yàn)證這一說(shuō)法，可以在”bats := batch.New(in, 5)“前面增加”time.Sleep(10* time.Millisecond)“讓隊(duì)列里數(shù)據(jù)有足夠數(shù)據(jù)，這時(shí)打印就會(huì)符合預(yù)期。

這里的batchWorker只有1個(gè)，也就是說(shuō)是向中心是串行上報(bào)，這個(gè)可以根據(jù)實(shí)際需求修改，通過(guò)開(kāi)啟多個(gè)協(xié)程來(lái)提高上報(bào)速度。

但是，有時(shí)候反而不滿足需求，這樣做會(huì)并發(fā)上報(bào)數(shù)據(jù)，增加中心壓力，同時(shí)由于消費(fèi)過(guò)快，反而使隊(duì)列中的數(shù)量就不容易達(dá)到分批數(shù)量，分批效果就減少，只是加快隊(duì)列消費(fèi)而已。

總結(jié)

Go的chan功能非常強(qiáng)大，各種并發(fā)處理都能看到他的身影，也是Go推薦的方式：

Do not communicate by sharing memory; instead, share memory by communicating.

靈活借助chan并發(fā)安全，既能阻塞又能select default非阻塞的特性，可以優(yōu)雅實(shí)現(xiàn)很多并發(fā)控制的需求

我的博客：Go實(shí)現(xiàn)并發(fā)扇入，批量扇出功能 | 藝術(shù)碼農(nóng)的小棧：https://itart.cn/blogs/2022/practice/go-chan-aggregate-batch.html