我希望在Go语言中构建一个缓冲区,支持多个并发读取器和一个写入器。任何写入到缓冲区的内容都应该被所有读取器读取。新的读取器可以随时加入,这意味着已经写入的数据必须能够为迟到的读取器回放。
缓冲区应满足以下接口:
type MyBuffer interface {
Write(p []byte) (n int, err error)
NextReader() io.Reader
}
你有没有建议,最好使用内置类型来实现这样的功能?
我希望在Go语言中构建一个缓冲区,支持多个并发读取器和一个写入器。任何写入到缓冲区的内容都应该被所有读取器读取。新的读取器可以随时加入,这意味着已经写入的数据必须能够为迟到的读取器回放。
缓冲区应满足以下接口:
type MyBuffer interface {
Write(p []byte) (n int, err error)
NextReader() io.Reader
}
你有没有建议,最好使用内置类型来实现这样的功能?
bufio.Scanner
)在遇到`io.EOF`后可能不会再读取更多数据(但这不是我们实现的缺陷)。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
,所以可以使用类型断言来"访问"其他os.File
的公开方法,即使它们不是MyBuf
接口的一部分。我认为这不是一个缺陷,但如果您想禁止这样做,您必须更改mybuf
的实现方式,不再嵌入os.File
,而是将其作为命名字段(但然后您必须自己添加Write()
和Close()
方法,正确地转发到os.File
字段)。
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()
中。io.WriteCloser
建议是个好主意。理想情况下,读取器的 Read()
方法会等待更多数据,直到写入器被关闭,此时它们将收到 io.EOF
。关于内存消耗,可以在内存中启动缓冲区,并在超过一定大小时将数据转储到磁盘上。我会采用您的建议来逐步解决这个问题。再次感谢您。 - Tympanix作为实验的一部分,我必须做类似的事情,所以分享:
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
}
这个偏移量由消费者控制:通常情况下,当消费者读取记录时,它会线性地推进其偏移量,但实际上,由于位置是由消费者控制的,因此它可以按任何顺序消耗记录。例如,消费者可以将偏移量重置为旧的位置以重新处理过去的数据,或者跳到最新的记录并从“现在”开始消费。
这些特性的组合意味着Kafka消费者非常便宜 - 它们可以来去自如,对集群或其他消费者的影响不大。例如,您可以使用我们的命令行工具“tail”任何主题的内容,而不会改变任何现有消费者所消费的内容。