带缓冲区和超时的Go编写器

mfuanj7w 于 2023-05-11 发布在 Go

关注(0)|答案(2)|浏览(107)

我想使用Go从SQS向AWS发送请求。有一些在单事件模式下使用v2 SDK中的sqs.SendMessageInput执行此操作的示例，但我希望改为批量发送。我已经创建了一个接口来从代码的其余部分中抽象出实现细节，沿着所示：

type UserRepository interface {
   Save(context.Context, User) error
}

如您所见，该接口不是特定于SQS的，可以很容易地替换为例如a Postgres实现。我真的很想保持界面尽可能干净。
使用SQS有一些我能想到的注意事项，可能也有一些我没有想到的。需要发送一个批次：
1.在每n次Save之后
1.在可配置的超时之后
1.上下文取消后
1.调用站点保存完所有User对象后
请注意，与发送到SQS不同，这也应该类似地用于写入控制台，这是我在示例中创建的。
这种设计是可能的，还是我必须在我的接口中始终有一个Close函数？
代码的问题是最后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")
}

来源：https://stackoverflow.com/questions/76189309/go-writer-with-buffer-and-timeout

2条答案

按热度按时间

ve7v8dk21#

有多种方法可以做到这一点。一般来说，不建议基于超时刷新此类缓冲实现，因为您无法控制如果失败会发生什么。
一种方法是进行显式批处理操作：

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
}

赞(0）回复(0）举报 2023-05-11

gk7wooem2#

UserRepository接口可以支持批处理操作而不需要关闭函数吗？
是的，一种解决方案是在界面中使用不同的Flush函数，调用者可以使用该函数显式刷新任何剩余的缓冲项。您的界面：

type UserRepository interface {
    Save(context.Context, User) error
    Flush(context.Context) error
    SaveAll(context.Context) error
}

根据您的喜好实现此UserRepository的一个示例可能是playground：

// for SO testing, by JS

package main

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

type User struct {
    ID   int
    Name string
}
type UserRepository interface {
    Save(context.Context, User) error
    Flush(context.Context) error
    SaveAll(context.Context) error
}

type LoggingBufferedUserRepository struct {
    buffer        []string
    bufferSize    int
    bufferTimeout time.Duration
    mutex         sync.Mutex
    // we don't need close here now
}

// add the item to the buffer and flushes the buffer if it reaches the configured size
func (c *LoggingBufferedUserRepository) Save(ctx context.Context, user User) error {
    // implement save functionality
    fmt.Printf("Save called %v: %v\n", user.Name, user.ID)
    return nil
}

// can - Flush function immediately flushes any buffered items
func (c *LoggingBufferedUserRepository) Flush(ctx context.Context) error {
    // implement Flush functionality
    fmt.Printf("Flush called\n")
    return nil
}

// waits for any pending items to be saved before returning, and flushes any remaining buffered items if there are any
func (c *LoggingBufferedUserRepository) SaveAll(ctx context.Context) error {
    // implement SaveAll functionality
    fmt.Printf("SaveAll called\n")
    return nil
}

func (c *LoggingBufferedUserRepository) flush(ctx context.Context) error {
    if len(c.buffer) == 0 {
        return nil
    }

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

    return nil
}

func main() {
    repo := &LoggingBufferedUserRepository{
        bufferSize:    5,
        bufferTimeout: 1 * time.Second,
    }

    users := []User{
        {ID: 1, Name: "Alice"},
        {ID: 2, Name: "Bob"},
        {ID: 3, Name: "Charlie"},
        {ID: 4, Name: "Dave"},
        {ID: 5, Name: "Eve"},
    }

    for _, user := range users {
        if err := repo.Save(context.Background(), user); err != nil {
            fmt.Printf("error saving user %v: %v\n", user, err)
        }
    }

    fmt.Printf("Waiting for flushing... (5 seconds)\n")
    time.Sleep(5 * time.Second)

    // Save remaining items
    if err := repo.SaveAll(context.Background()); err != nil {
        fmt.Printf("error saving remaining items: %v\n", err)
    }

    // Flush remaining items
    if err := repo.Flush(context.Background()); err != nil {
        fmt.Printf("error flushing remaining items: %v\n", err)
    }
}

其输出应为：

Save called Alice: 1
Save called Bob: 2
Save called Charlie: 3
Save called Dave: 4
Save called Eve: 5
Waiting for flushing... (5 seconds)
SaveAll called
Flush called

Program exited.

赞(0）回复(0）举报 2023-05-11

我来回答

带缓冲区和超时的Go编写器

2条答案

相关问题

热门标签

最新问答