包含缓冲区与超时机制的Go写入器

Golang不知道大家是否熟悉？今天我将给大家介绍《包含缓冲区与超时机制的Go写入器》，这篇文章主要会讲到等等知识点，如果你在看完本篇文章后，有更好的建议或者发现哪里有问题，希望大家都能积极评论指出，谢谢！希望我们能一起加油进步！

我想使用 go 从 sqs 向 aws 发送请求。有使用 v2 sdk 中的 sqs.sendmessageinput 在单事件模式下执行此操作的示例，但我想改为批量发送。我创建了一个接口来从代码的其余部分中抽象出实现细节，大致如下：

type userrepository interface {
   save(context.context, user) error
}

正如您所看到的，该接口不是 sqs 特定的，可以很容易地被替换，例如postgres 实现。我真的很想保持界面尽可能干净。

使用 sqs 有一些我能想到的注意事项，也可能有一些我没有想到的注意事项。 需要发送一批：

1. 每次 nth save 后
2. 经过可配置的超时
3. 上下文取消后
4. 调用站点完成保存所有 user 对象后

请注意，与发送到 sqs 不同，这也应该适用于写入控制台，这就是我在示例中创建的内容。

这种设计是否可行，还是我的界面中始终必须有 clos​​e 函数？

代码的问题是最后 4 个 (nr_of_items % batch_size) 将不会被“保存”。

package main

import (
    "context"
    "fmt"
    "os"
    "os/signal"
    "sync"
    "time"
)

type LoggingBufferedUserRepository struct {
    buffer        []string
    bufferSize    int
    bufferTimeout time.Duration
    mutex         sync.Mutex
    closeChan     chan struct{}
}

func NewLoggingBufferedUserRepository(
    ctx context.Context, bufferSize int, bufferTimeout time.Duration,
) *LoggingBufferedUserRepository {
    client := &LoggingBufferedUserRepository{
        bufferSize:    bufferSize,
        bufferTimeout: bufferTimeout,
        closeChan:     make(chan struct{}),
    }

    go client.bufferMonitor(ctx)
    return client
}

func (c *LoggingBufferedUserRepository) SendMessage(ctx context.Context, input string) {
    c.mutex.Lock()
    defer c.mutex.Unlock()
    c.buffer = append(c.buffer, input)
    if len(c.buffer) >= c.bufferSize {
        go c.flush(ctx, c.buffer)
        c.buffer = []string{}
    }
    return
}

func (c *LoggingBufferedUserRepository) flush(ctx context.Context, buffer []string) {
    if len(buffer) == 0 {
        return
    }

    // This is the actual batch 'save':
    fmt.Printf("flushing buffer, size=%d, cid=%s, buffer=$%v\n", len(buffer), ctx.Value("cid"), buffer)
}

func (c *LoggingBufferedUserRepository) bufferMonitor(ctx context.Context) {
    timeout := time.NewTimer(c.bufferTimeout)
    for {
        select {
        case <-timeout.C:
            c.flush(ctx, c.buffer)
            c.buffer = []string{}
        }

        timeout.Reset(c.bufferTimeout)
    }
}

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
    defer stop()
    g := NewLoggingBufferedUserRepository(ctx, 10, 1*time.Second)

    wg := &sync.WaitGroup{}
    for i := 0; i < 15; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            fmt.Printf("sending message %d\n", i)
            g.SendMessage(ctx, fmt.Sprintf("a%d", i))
        }(i)
        time.Sleep(100 * time.Millisecond)
    }
    wg.Wait()
    fmt.Println("done")
}

正确答案

有多种方法可以做到这一点。一般来说，不建议根据超时刷新此类缓冲实现，因为您无法控制失败时会发生什么。

一种方法是进行显式批处理操作：

type userrepository interface {
   // save single user
   save(context.context, user) error
   // save batch
   savebatch(context.context) userbatch
}

哪里

type UserBatch interface {
   Save(context.Context,User) error
   Flush(context.Context) error
   Close(context.Context) error
}

以上就是《包含缓冲区与超时机制的Go写入器》的详细内容，更多关于的资料请关注golang学习网公众号！