Golang缓冲区与并发读者

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Golang缓冲区与并发读者相关的知识,希望对你有一定的参考价值。

我想在Go中构建一个缓冲区,它支持多个并发读者和一个编写器。所有读者都应该阅读写入缓冲区的内容。允许新读者随时进入,这意味着已经写好的数据必须能够为后期读者播放。

缓冲区应满足以下接口:

type MyBuffer interface {
    Write(p []byte) (n int, err error)
    NextReader() io.Reader
}

对于这样的实现,您是否有任何建议,最好使用内置类型?

答案

根据本作者的性质以及如何使用它,将所有内容保存在内存中(以便能够重新播放所有内容供读者加入)是非常危险的,并且可能需要大量内存,或者导致应用程序崩溃内存不足。

将它用于“低流量”记录器可以将所有内容保存在内存中可能没问题,但是例如流式传输某些音频或视频很可能不是。

如果下面的读者实现读取了写入缓冲区的所有数据,他们的Read()方法将正确报告io.EOF。必须小心,因为一旦遇到bufio.Scanner,一些构造(例如io.EOF)可能无法读取更多数据(但这不是我们实现的缺陷)。

如果你希望缓冲区的读者等待缓冲区中没有更多数据可用,要等到写入新数据而不是返回io.EOF,你可以将返回的读者包装在这里提供的“尾读者”中:Go: "tail -f"-like generator

"Memory-safe" file implementation

这是一个非常简单和优雅的解决方案。它使用文件写入,并使用文件进行读取。同步基本上由操作系统提供。这不存在内存不足错误的风险,因为数据仅存储在磁盘上。根据作者的性质,这可能是也可能不够。

我宁愿使用以下界面,因为Close()在文件的情况下很重要。

type MyBuf interface {
    io.WriteCloser
    NewReader() (io.ReadCloser, error)
}

而且实现非常简单:

type mybuf struct {
    *os.File
}

func (mb *mybuf) NewReader() (io.ReadCloser, error) {
    f, err := os.Open(mb.Name())
    if err != nil {
        return nil, err
    }
    return f, nil
}

func NewMyBuf(name string) (MyBuf, error) {
    f, err := os.Create(name)
    if err != nil {
        return nil, err
    }
    return &mybuf{File: f}, nil
}

我们的mybuf类型嵌入*os.File,所以我们得到Write()Close()方法为“免费”。

NewReader()只是打开现有的后备文件进行读取(以只读模式)并返回它,再次利用它实现io.ReadCloser

创建一个新的MyBuf值正在NewMyBuf()函数中实现,如果创建文件失败,它也可能返回error

笔记:

请注意,由于mybuf嵌入*os.File,有可能使用type assertion“到达”os.File的其他导出方法,即使它们不是MyBuf界面的一部分。我不认为这是一个缺陷,但如果你想禁止这个,你必须改变mybuf的实现,不要嵌入os.File,而是将它作为命名字段(但是你必须自己添加Write()Close()方法,正确转发到os.File领域)。

In-memory implementation

如果文件实现不充分,那么就会出现内存实现。

由于我们现在只在内存中,我们将使用以下界面:

type MyBuf interface {
    io.Writer
    NewReader() io.Reader
}

我们的想法是存储所有传递给缓冲区的字节片。当调用Read()时,读者将提供存储的切片,每个读者将跟踪其Read()方法服务的存储切片的数量。必须处理同步,我们将使用简单的sync.RWMutex

不用多说,这是实现:

type mybuf struct {
    data [][]byte
    sync.RWMutex
}

func (mb *mybuf) Write(p []byte) (n int, err error) {
    if len(p) == 0 {
        return 0, nil
    }
    // Cannot retain p, so we must copy it:
    p2 := make([]byte, len(p))
    copy(p2, p)
    mb.Lock()
    mb.data = append(mb.data, p2)
    mb.Unlock()
    return len(p), nil
}

type mybufReader struct {
    mb   *mybuf // buffer we read from
    i    int    // next slice index
    data []byte // current data slice to serve
}

func (mbr *mybufReader) Read(p []byte) (n int, err error) {
    if len(p) == 0 {
        return 0, nil
    }
    // Do we have data to send?
    if len(mbr.data) == 0 {
        mb := mbr.mb
        mb.RLock()
        if mbr.i < len(mb.data) {
            mbr.data = mb.data[mbr.i]
            mbr.i++
        }
        mb.RUnlock()
    }
    if len(mbr.data) == 0 {
        return 0, io.EOF
    }

    n = copy(p, mbr.data)
    mbr.data = mbr.data[n:]
    return n, nil
}

func (mb *mybuf) NewReader() io.Reader {
    return &mybufReader{mb: mb}
}

func NewMyBuf() MyBuf {
    return &mybuf{}
}

请注意,Writer.Write()的一般合同包括实现不能保留传递的切片,因此我们必须在“存储”之前复制它。

另请注意,Read()的读者试图锁定最少的时间。也就是说,它只在我们需要缓冲区中的新数据切片时锁定,并且只进行读锁定,这意味着如果读取器具有部分数据切片,则将在Read()中发送该数据切片而不锁定并触摸缓冲区。

另一答案

我链接到仅附加提交日志,因为它看起来非常类似于您的要求。我对分布式系统和提交日志都很陌生,所以我可能会对几个概念进行屠宰,但是kafka的介绍清楚地解释了所有的图表。

Go对我来说也很新鲜,所以我确信有更好的方法:

但也许您可以将缓冲区建模为切片,我认为有几种情况:

  • 缓冲区没有读取器,新数据被写入缓冲区,缓冲区长度增加
  • 缓冲区有一个/多个读卡器: 读者订阅缓冲区 buffer创建并向该客户端返回一个通道 buffer维护一个客户端通道列表 写入发生 - >循环遍历所有客户端通道并发布到它(pub sub)

这解决了pubsub实时消费者流,其中消息被扇出,但没有解决回填问题。

Kafka使回填和他们的intro illustrates如何做到:)

这种偏移由消费者控制:通常消费者在读取记录时会线性地提高其偏移量,但事实上,由于消费者控制位置,它可以按照自己喜欢的任何顺序消费记录。例如,消费者可以重置为较旧的偏移量以重新处理过去的数据,或者跳到最近的记录并从“现在”开始消费。

这些功能组合意味着Kafka消费者非常便宜 - 他们可以来来往往对集群或其他消费者没有太大影响。例如,您可以使用我们的命令行工具“拖尾”任何主题的内容,而无需更改任何现有使用者所消耗的内容。

另一答案

作为实验的一部分,我必须做类似的事情,所以分享:

type MultiReaderBuffer struct {
    mu  sync.RWMutex
    buf []byte
}

func (b *MultiReaderBuffer) Write(p []byte) (n int, err error) {
    if len(p) == 0 {
        return 0, nil
    }
    b.mu.Lock()
    b.buf = append(b.buf, p...)
    b.mu.Unlock()
    return len(p), nil
}

func (b *MultiReaderBuffer) NewReader() io.Reader {
    return &mrbReader{mrb: b}
}

type mrbReader struct {
    mrb *MultiReaderBuffer
    off int
}

func (r *mrbReader) Read(p []byte) (n int, err error) {
    if len(p) == 0 {
        return 0, nil
    }
    r.mrb.mu.RLock()
    n = copy(p, r.mrb.buf[r.off:])
    r.mrb.mu.RUnlock()
    if n == 0 {
        return 0, io.EOF
    }
    r.off += n
    return n, nil
}

以上是关于Golang缓冲区与并发读者的主要内容,如果未能解决你的问题,请参考以下文章

golang代码片段(摘抄)

golang的缓冲channel和无缓冲channel的区别

Golang入门到项目实战 golang并发变成之通道channel

golang net/http包 http请求的字节码读取与解析。

[Go] 通过 17 个简短代码片段,切底弄懂 channel 基础

用生产者消费者理解golang channel