如果通道已满或在特定时间后从 Go 缓冲通道读取

来到golang学习网的大家，相信都是编程学习爱好者，希望在这里学习Golang相关编程知识。下面本篇文章就来带大家聊聊《如果通道已满或在特定时间后从 Go 缓冲通道读取》，介绍一下，希望对大家的知识积累有所帮助，助力实战开发！

如果通道已满或经过一定时间，我想从通道进行处理。我的用例与现有问题类似，我尝试修改现有答案

我的代码位于 https://go.dev/play/p/hagz9hhqj0i，

package main

import (
    "fmt"
    "sync"
    "time"
)

type audit struct {
    id int
}

const batchsize = 5

var batch = make([]audit, 0, batchsize)

func upsertbigqueryaudits(audits []audit) {
    fmt.printf("processing batch of %d\n", len(audits))
    for _, a := range audits {
        fmt.printf("%d ", a.id)
    }
    fmt.println()
    batch = []audit{}
}
func process(full <-chan struct{}) {
    ticker := time.newticker(1 * time.nanosecond)

    for {
        select {
        case <-full:
            fmt.println("from full")
            upsertbigqueryaudits(batch)
        case <-ticker.c:
            fmt.println("from ticker")
            if len(batch) > 0 {
                fmt.println("from ticker sending batch")
                upsertbigqueryaudits(batch)
            }
        }
    }
}
func processaudits(audits <-chan audit, full chan<- struct{}, batchsize int) {
    for audit := range audits {
        batch = append(batch, audit)
        if len(batch) == cap(batch) {
            // upsertbigqueryaudits(batch)
            fmt.println("sending full")
            full <- struct{}{}

        }
    }

}

func produceaudits(x int, to chan audit) {
    for i := 0; i < x; i++ {
        to <- audit{
            id: i,
        }
    }
}

func main() {
    var wg sync.waitgroup
    audits := make(chan audit)
    full := make(chan struct{})
    wg.add(1)
    go func() {
        defer wg.done()
        process(full)
    }()
    wg.add(1)
    go func() {
        defer wg.done()
        processaudits(audits, full, batchsize)
    }()
    wg.add(1)
    go func() {
        defer wg.done()
        produceaudits(25, audits)
        close(audits)
    }()
    wg.wait()
    fmt.println("complete")
}

在这里，我想在批处理充满 batchsize 元素时或在经过一定时间（例如 1 纳秒）时处理批处理。

编辑：内联添加了我的担忧。 这是示例输出。

timeout running program
Sending full <--- this line comes twice
Sending full
From full
Processing batch of 10 <--- instead of 5 it is processing 10 items
0 1 2 3 4 5 6 7 8 9 
From full
Processing batch of 1
10 
Sending full
Sending full
From full
Processing batch of 1 <-- instead of 5 it is processing 1 items. It does not process more than 1 item.
11 
From full
Processing batch of 0

...
...
Sending full
Sending full
From full
Processing batch of 1 <-- instead of 5 it is processing 1 items
24 
From full
Processing batch of 0

From ticker <-- this line comes only after consuming 25 items.  
From ticker
From ticker
From ticker

正确答案

我在https://elliotchance.medium.com/batch-a-channel-by-size-or-time-in-go-92fa3098f65找到了答案

package main

import (
    "context"
    "fmt"
    "time"
)

func BatchStringsCtx[T any](ctx context.Context, values <-chan T, maxItems int, maxTimeout time.Duration) chan []T {
    batches := make(chan []T)

    go func() {
        defer close(batches)

        for keepGoing := true; keepGoing; {
            var batch []T
            expire := time.After(maxTimeout)
            for {
                select {
                case <-ctx.Done():
                    keepGoing = false
                    goto done

                case value, ok := <-values:
                    if !ok {
                        keepGoing = false
                        goto done
                    }

                    batch = append(batch, value)
                    if len(batch) == maxItems {
                        goto done
                    }

                case <-expire:
                    goto done
                }
            }

        done:
            if len(batch) > 0 {
                batches <- batch
            }
        }
    }()

    return batches
}

func main() {
    strings := make(chan string)
    go func() {
        strings <- "hello"
        strings <- "there" // hit limit of 2

        strings <- "how"
        time.Sleep(15 * time.Millisecond) // hit timeout

        strings <- "are"
        strings <- "you" // hit limit of 2

        // A really long time without any new values.
        // The context was cancelled around 300ms,
        // before this sleep finished.
        time.Sleep(500 * time.Millisecond)

        strings <- "doing?" // never read

        close(strings)
    }()

    //ctx, cancel := context.WithTimeout(context.Background(), 300 * time.Millisecond)
    ctx := context.Background()

    start := time.Now()
    batches := BatchStringsCtx[string](ctx, strings, 2, 10*time.Millisecond)
    for batch := range batches {
        fmt.Println(time.Now().Sub(start), batch)
    }
}

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~