758 lines
20 KiB
Go
758 lines
20 KiB
Go
/**
|
||
* 基于磁盘的 FIFO 先进先出消息队列
|
||
* source nsq
|
||
* https://github.com/nsqio/go-diskqueue/blob/master/diskqueue.go
|
||
*/
|
||
|
||
package disk
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"encoding/binary"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"math/rand"
|
||
"os"
|
||
"path"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
type LogLevel int
|
||
|
||
const (
|
||
DEBUG = LogLevel(1)
|
||
INFO = LogLevel(2)
|
||
WARN = LogLevel(3)
|
||
ERROR = LogLevel(4)
|
||
FATAL = LogLevel(5)
|
||
)
|
||
|
||
type AppLogFunc func(lvl LogLevel, f string, args ...interface{})
|
||
|
||
func (l LogLevel) String() string {
|
||
switch l {
|
||
case 1:
|
||
return "DEBUG"
|
||
case 2:
|
||
return "INFO"
|
||
case 3:
|
||
return "WARNING"
|
||
case 4:
|
||
return "ERROR"
|
||
case 5:
|
||
return "FATAL"
|
||
}
|
||
panic("invalid LogLevel")
|
||
}
|
||
|
||
type IDiskQueue interface {
|
||
// Put 加锁检测 标志位是否退出, 如果否 则继续往文件写入数据并等待结果
|
||
// 写入*.meta.dat 格式
|
||
// 1.depth
|
||
// 2.readFileNum,readPos
|
||
// 3.writeFileNum,writePos
|
||
Put([]byte) error
|
||
// ReadChan 读取文件数据 返回只读的chan 可用于多消费者并发读取
|
||
ReadChan() <-chan []byte
|
||
PeekChan() <-chan []byte
|
||
// Close 等待loop结束, 正常关闭 并保存元数据
|
||
Close() error
|
||
// Delete 等待loop结束, 直接关闭io流
|
||
Delete() error
|
||
// Depth 未读消息积压量
|
||
Depth() int64
|
||
// Empty 清空消息, 删除文件
|
||
Empty() error
|
||
}
|
||
|
||
// diskQueue 基于文件系统的先进先出队列
|
||
type diskQueue struct {
|
||
// 64bit atomic vars need to be first for proper alignment on 32bit platforms
|
||
|
||
// run-time state (also persisted to disk)
|
||
// 运行时的数据保存, 也会保存到文件
|
||
readPos int64 // 文件读取的位置
|
||
writePos int64 // 文件写入的位置
|
||
readFileNum int64 // 读取文件编号
|
||
writeFileNum int64 // 写入文件编号
|
||
depth int64 // 未读消息积压量
|
||
|
||
sync.RWMutex // 读写锁
|
||
|
||
// instantiation time metadata
|
||
name string // 队列实例名称
|
||
dataPath string // 数据文件存储路径
|
||
maxBytesPerFile int64 // 每个文件最大字节数
|
||
maxBytesPerFileRead int64
|
||
minMsgSize int32 // 最小消息长度
|
||
maxMsgSize int32 // 最大消息长度
|
||
syncEvery int64 // 刷盘频率设置
|
||
syncTimeout time.Duration // 刷盘时间设置（单位秒）
|
||
exitFlag int32 // 退出标志位
|
||
needSync bool // 是否需要同步
|
||
|
||
// 跟踪读取的位置
|
||
nextReadPos int64 // 下一次读取的位置
|
||
nextReadFileNum int64 // 下一次读取对应的文件编号
|
||
|
||
readFile *os.File // 读取的文件
|
||
writeFile *os.File // 写入的文件
|
||
reader *bufio.Reader // 缓冲读取
|
||
writeBuf bytes.Buffer // 缓冲写入
|
||
|
||
// 通过 ReadChan() 方法对外暴露
|
||
readChan chan []byte // 读取的数据,可以多消费者进行通信消费
|
||
|
||
// 通过 PeekChan() 方法对外暴露
|
||
peekChan chan []byte // 探查数据
|
||
|
||
// 内部管道
|
||
depthChan chan int64
|
||
writeChan chan []byte // 写入通道
|
||
writeResponseChan chan error // 写入结果反馈通道
|
||
emptyChan chan int // 清空队列通道
|
||
emptyResponseChan chan error // 清空反馈通道
|
||
exitChan chan int // 结束信号通道
|
||
exitSyncChan chan int // 退出同步通道
|
||
logf AppLogFunc // 日志记录封装
|
||
}
|
||
|
||
// NewDiskQueue 实例化一个diskQueue，从文件系统中检索元数据并启动预计程序
|
||
func NewDiskQueue(name, dataPath string, maxBytesPerFile int64, minMsgSize, maxMsgSize int32, syncEvery int64, syncTimeout time.Duration, logf AppLogFunc) IDiskQueue {
|
||
d := diskQueue{
|
||
name: name,
|
||
dataPath: dataPath,
|
||
maxBytesPerFile: maxBytesPerFile,
|
||
minMsgSize: minMsgSize,
|
||
maxMsgSize: maxMsgSize,
|
||
syncEvery: syncEvery,
|
||
syncTimeout: syncTimeout,
|
||
readChan: make(chan []byte),
|
||
peekChan: make(chan []byte),
|
||
depthChan: make(chan int64),
|
||
writeChan: make(chan []byte),
|
||
writeResponseChan: make(chan error),
|
||
emptyChan: make(chan int),
|
||
emptyResponseChan: make(chan error),
|
||
exitChan: make(chan int),
|
||
exitSyncChan: make(chan int),
|
||
logf: logf,
|
||
}
|
||
// 从磁盘加载元数据
|
||
err := d.retrieveMetaData()
|
||
if err != nil && !os.IsNotExist(err) {
|
||
d.logf(ERROR, "DISKQUEUE(%s) failed to retrieveMetaData - %s", d.name, err)
|
||
}
|
||
go d.ioLoop()
|
||
return &d
|
||
}
|
||
|
||
// Depth 获取消息深度(条数)
|
||
func (d *diskQueue) Depth() int64 {
|
||
depth, ok := <-d.depthChan
|
||
if !ok {
|
||
// ioLoop exited
|
||
depth = d.depth
|
||
}
|
||
return depth
|
||
}
|
||
|
||
// ReadChan 返回一个用于读取字节型数据的通道
|
||
func (d *diskQueue) ReadChan() <-chan []byte {
|
||
return d.readChan
|
||
}
|
||
|
||
// PeekChan 返回一个用于探测字数据的通道，不会对影响消息队列状态
|
||
func (d *diskQueue) PeekChan() <-chan []byte {
|
||
return d.peekChan
|
||
}
|
||
|
||
// Put 写入字节型数据到队列中
|
||
func (d *diskQueue) Put(data []byte) error {
|
||
d.RLock()
|
||
defer d.RUnlock()
|
||
|
||
if d.exitFlag == 1 {
|
||
return errors.New("exiting")
|
||
}
|
||
|
||
d.writeChan <- data
|
||
return <-d.writeResponseChan
|
||
}
|
||
|
||
// Close cleans up the queue and persists metadata
|
||
// 清理队列并持久化元数据
|
||
func (d *diskQueue) Close() error {
|
||
err := d.exit(false)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return d.sync()
|
||
}
|
||
|
||
func (d *diskQueue) Delete() error {
|
||
return d.exit(true)
|
||
}
|
||
|
||
func (d *diskQueue) exit(deleted bool) error {
|
||
d.Lock()
|
||
defer d.Unlock()
|
||
|
||
d.exitFlag = 1
|
||
|
||
if deleted {
|
||
d.logf(INFO, "DISKQUEUE(%s): deleting", d.name)
|
||
} else {
|
||
d.logf(INFO, "DISKQUEUE(%s): closing", d.name)
|
||
}
|
||
// 关闭退出通道
|
||
close(d.exitChan)
|
||
|
||
// 确保ioLoop已经退出
|
||
<-d.exitSyncChan
|
||
|
||
// 关闭文件读写
|
||
close(d.depthChan)
|
||
if d.readFile != nil {
|
||
d.readFile.Close()
|
||
d.readFile = nil
|
||
}
|
||
if d.writeFile != nil {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// Empty destructively clears out any pending data in the queue
|
||
// by fast forwarding read positions and removing intermediate files
|
||
// 破坏性的清空数据，删除文件
|
||
func (d *diskQueue) Empty() error {
|
||
d.RLock()
|
||
defer d.RUnlock()
|
||
|
||
if d.exitFlag == 1 {
|
||
return errors.New("exiting")
|
||
}
|
||
|
||
d.logf(INFO, "DISKQUEUE(%s): emptying", d.name)
|
||
|
||
d.emptyChan <- 1
|
||
return <-d.emptyResponseChan
|
||
}
|
||
|
||
// 删除所有文件
|
||
func (d *diskQueue) deleteAllFiles() error {
|
||
err := d.skipToNextRWFile()
|
||
|
||
innerErr := os.Remove(d.metaDataFileName())
|
||
if innerErr != nil && !os.IsNotExist(innerErr) {
|
||
d.logf(ERROR, "DISKQUEUE(%s) failed to remove metadata file - %s", d.name, innerErr)
|
||
return innerErr
|
||
}
|
||
|
||
return err
|
||
}
|
||
|
||
// 跳到下一个读写文件
|
||
func (d *diskQueue) skipToNextRWFile() error {
|
||
var err error
|
||
|
||
if d.readFile != nil {
|
||
d.readFile.Close()
|
||
d.readFile = nil
|
||
}
|
||
|
||
if d.writeFile != nil {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
}
|
||
|
||
for i := d.readFileNum; i <= d.writeFileNum; i++ {
|
||
fileName := d.fileName(i)
|
||
innerErr := os.Remove(fileName)
|
||
if innerErr != nil && !os.IsNotExist(innerErr) {
|
||
d.logf(ERROR, "DISKQUEUE(%s) failed to remove data file - %s", d.name, innerErr)
|
||
err = innerErr
|
||
}
|
||
}
|
||
|
||
d.writeFileNum++
|
||
d.writePos = 0
|
||
d.readFileNum = d.writeFileNum
|
||
d.readPos = 0
|
||
d.nextReadFileNum = d.writeFileNum
|
||
d.nextReadPos = 0
|
||
d.depth = 0
|
||
|
||
return err
|
||
}
|
||
|
||
func (d *diskQueue) readOne() ([]byte, error) {
|
||
var err error
|
||
var msgSize int32
|
||
|
||
if d.readFile == nil {
|
||
curFileName := d.fileName(d.readFileNum)
|
||
d.readFile, err = os.OpenFile(curFileName, os.O_RDONLY, 0600)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
d.logf(INFO, "DISKQUEUE(%s): readOne() opened %s", d.name, curFileName)
|
||
|
||
if d.readPos > 0 {
|
||
// 设置读取偏移
|
||
_, err = d.readFile.Seek(d.readPos, 0)
|
||
if err != nil {
|
||
d.readFile.Close()
|
||
d.readFile = nil
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
// for "complete" files (i.e. not the "current" file), maxBytesPerFileRead
|
||
// should be initialized to the file's size, or default to maxBytesPerFile
|
||
// maxBytesPerFileRead应该初始化为文件的大小，或者默认为maxBytesPerFile
|
||
d.maxBytesPerFileRead = d.maxBytesPerFile
|
||
if d.readFileNum < d.writeFileNum {
|
||
stat, err := d.readFile.Stat()
|
||
if err == nil {
|
||
d.maxBytesPerFileRead = stat.Size()
|
||
}
|
||
}
|
||
|
||
d.reader = bufio.NewReader(d.readFile)
|
||
}
|
||
|
||
err = binary.Read(d.reader, binary.BigEndian, &msgSize)
|
||
if err != nil {
|
||
d.readFile.Close()
|
||
d.readFile = nil
|
||
return nil, err
|
||
}
|
||
|
||
if msgSize < d.minMsgSize || msgSize > d.maxMsgSize {
|
||
// this file is corrupt and we have no reasonable guarantee on
|
||
// where a new message should begin
|
||
// 文件损坏，无法读取
|
||
d.readFile.Close()
|
||
d.readFile = nil
|
||
return nil, fmt.Errorf("invalid message read size (%d)", msgSize)
|
||
}
|
||
|
||
readBuf := make([]byte, msgSize)
|
||
_, err = io.ReadFull(d.reader, readBuf)
|
||
if err != nil {
|
||
d.readFile.Close()
|
||
d.readFile = nil
|
||
return nil, err
|
||
}
|
||
|
||
totalBytes := int64(4 + msgSize)
|
||
|
||
// we only advance next* because we have not yet sent this to consumers
|
||
// (where readFileNum, readPos will actually be advanced)
|
||
d.nextReadPos = d.readPos + totalBytes
|
||
d.nextReadFileNum = d.readFileNum
|
||
|
||
// we only consider rotating if we're reading a "complete" file
|
||
// and since we cannot know the size at which it was rotated, we
|
||
// rely on maxBytesPerFileRead rather than maxBytesPerFile
|
||
// if d.readFileNum < d.writeFileNum && d.nextReadPos >= d.maxBytesPerFileRead {
|
||
if d.readFileNum <= d.writeFileNum && d.nextReadPos >= d.maxBytesPerFileRead {
|
||
if d.readFile != nil {
|
||
d.readFile.Close()
|
||
d.readFile = nil
|
||
}
|
||
|
||
d.nextReadFileNum++
|
||
d.nextReadPos = 0
|
||
}
|
||
|
||
return readBuf, nil
|
||
}
|
||
|
||
// writeOne performs a low level filesystem write for a single []byte
|
||
// while advancing write positions and rolling files, if necessary
|
||
func (d *diskQueue) writeOne(data []byte) error {
|
||
var err error
|
||
|
||
dataLen := int32(len(data))
|
||
totalBytes := int64(4 + dataLen)
|
||
|
||
if dataLen < d.minMsgSize || dataLen > d.maxMsgSize {
|
||
return fmt.Errorf("invalid message write size (%d) minMsgSize=%d maxMsgSize=%d", dataLen, d.minMsgSize, d.maxMsgSize)
|
||
}
|
||
|
||
// will not wrap-around if maxBytesPerFile + maxMsgSize < Int64Max
|
||
// 如果文件大小超过了设置的最大值，关闭当前写入文件
|
||
if d.writePos > 0 && d.writePos+totalBytes > d.maxBytesPerFile {
|
||
if d.readFileNum == d.writeFileNum {
|
||
d.maxBytesPerFileRead = d.writePos
|
||
}
|
||
|
||
d.writeFileNum++
|
||
d.writePos = 0
|
||
|
||
// sync every time we start writing to a new file
|
||
err = d.sync()
|
||
if err != nil {
|
||
d.logf(ERROR, "DISKQUEUE(%s) failed to sync - %s", d.name, err)
|
||
}
|
||
|
||
if d.writeFile != nil {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
}
|
||
}
|
||
|
||
if d.writeFile == nil {
|
||
curFileName := d.fileName(d.writeFileNum)
|
||
d.writeFile, err = os.OpenFile(curFileName, os.O_RDWR|os.O_CREATE, 0600)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
d.logf(INFO, "DISKQUEUE(%s): writeOne() opened %s", d.name, curFileName)
|
||
|
||
if d.writePos > 0 {
|
||
_, err = d.writeFile.Seek(d.writePos, 0)
|
||
if err != nil {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
|
||
d.writeBuf.Reset()
|
||
// 缓冲区写入二进制数据
|
||
err = binary.Write(&d.writeBuf, binary.BigEndian, dataLen)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
_, err = d.writeBuf.Write(data)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// only write to the file once
|
||
// 写入到文件
|
||
_, err = d.writeFile.Write(d.writeBuf.Bytes())
|
||
if err != nil {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
return err
|
||
}
|
||
|
||
d.writePos += totalBytes
|
||
d.depth += 1
|
||
|
||
return err
|
||
}
|
||
|
||
// sync fsyncs the current writeFile and persists metadata
|
||
// 刷盘，持久化数据
|
||
func (d *diskQueue) sync() error {
|
||
if d.writeFile != nil {
|
||
// 数据写入磁盘
|
||
err := d.writeFile.Sync()
|
||
if err != nil {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
return err
|
||
}
|
||
}
|
||
|
||
// 持久化元数据
|
||
err := d.persistMetaData()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
d.needSync = false
|
||
return nil
|
||
}
|
||
|
||
// retrieveMetaData initializes state from the filesystem
|
||
// 从本地文件取回元数据
|
||
func (d *diskQueue) retrieveMetaData() error {
|
||
var f *os.File
|
||
var err error
|
||
|
||
// 存储路径.diskqueue.meta.dat
|
||
fileName := d.metaDataFileName()
|
||
f, err = os.OpenFile(fileName, os.O_RDONLY, 0600)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer f.Close()
|
||
|
||
var depth int64
|
||
_, err = fmt.Fscanf(f, "%d\n%d,%d\n%d,%d\n",
|
||
&depth, // 待读取消息数量
|
||
&d.readFileNum, // 待读取文件编号
|
||
&d.readPos, // 待读取文件位置
|
||
&d.writeFileNum, // 待写入文件编号
|
||
&d.writePos, // 待写入文件位置
|
||
)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
d.depth = depth
|
||
// 更新读取位置和文件编号
|
||
d.nextReadFileNum = d.readFileNum
|
||
d.nextReadPos = d.readPos
|
||
|
||
// 如果元数据在最后一次关闭系统时没有同步，
|
||
// 那么实际文件的大小实际上可能大于writePos写入位置，
|
||
// 在这种情况下，最安全的做法是跳到下一个文件进行写操作，
|
||
// 并让读取器从磁盘队列中的消息中尽可能地抢救出元数据(可能也是过时的readPos)之外的内容
|
||
fileName = d.fileName(d.writeFileNum)
|
||
fileInfo, err := os.Stat(fileName)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
fileSize := fileInfo.Size()
|
||
if d.writePos < fileSize {
|
||
d.logf(WARN, "DISKQUEUE(%s) %s metadata writePos %d < file size of %d, skipping to new file", d.name, fileName, d.writePos, fileSize)
|
||
d.writeFileNum += 1
|
||
d.writePos = 0
|
||
if d.writeFile != nil {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// persistMetaData atomically writes state to the filesystem
|
||
// 同步元数据到本地文件
|
||
func (d *diskQueue) persistMetaData() error {
|
||
var f *os.File
|
||
var err error
|
||
// metaDat 临时文件
|
||
fileName := d.metaDataFileName()
|
||
tmpFileName := fmt.Sprintf("%s.%d.tmp", fileName, rand.Int())
|
||
|
||
// 写入到临时文件
|
||
f, err = os.OpenFile(tmpFileName, os.O_RDWR|os.O_CREATE, 0600)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
_, err = fmt.Fprintf(f, "%d\n%d,%d\n%d,%d\n", d.depth, d.readFileNum, d.readPos, d.writeFileNum, d.writePos)
|
||
if err != nil {
|
||
f.Close()
|
||
return err
|
||
}
|
||
f.Sync()
|
||
f.Close()
|
||
// 成功往临时文件写入数据, 在进行替换源文件
|
||
// atomically rename
|
||
return os.Rename(tmpFileName, fileName)
|
||
}
|
||
|
||
// 获取元数据文件名称
|
||
func (d *diskQueue) metaDataFileName() string {
|
||
return fmt.Sprintf(path.Join(d.dataPath, "%s.diskqueue.meta.dat"), d.name)
|
||
}
|
||
|
||
// 获取文件名称
|
||
func (d *diskQueue) fileName(fileNum int64) string {
|
||
return fmt.Sprintf(path.Join(d.dataPath, "%s.diskqueue.%06d.dat"), d.name, fileNum)
|
||
}
|
||
|
||
func (d *diskQueue) checkTailCorruption(depth int64) {
|
||
if d.readFileNum < d.writeFileNum || d.readPos < d.writePos {
|
||
return
|
||
}
|
||
|
||
// we've reached the end of the diskqueue
|
||
// if depth isn't 0 something went wrong
|
||
if depth != 0 {
|
||
if depth < 0 {
|
||
d.logf(ERROR, "DISKQUEUE(%s) negative depth at tail (%d), metadata corruption, resetting 0...", d.name, depth)
|
||
} else if depth > 0 {
|
||
d.logf(ERROR, "DISKQUEUE(%s) positive depth at tail (%d), data loss, resetting 0...", d.name, depth)
|
||
}
|
||
// 强制设置为0
|
||
d.depth = 0
|
||
d.needSync = true
|
||
}
|
||
|
||
if d.readFileNum != d.writeFileNum || d.readPos != d.writePos {
|
||
if d.readFileNum > d.writeFileNum {
|
||
d.logf(ERROR, "DISKQUEUE(%s) readFileNum > writeFileNum (%d > %d), corruption, skipping to next writeFileNum and resetting 0...", d.name, d.readFileNum, d.writeFileNum)
|
||
}
|
||
if d.readPos > d.writePos {
|
||
d.logf(ERROR, "DISKQUEUE(%s) readPos > writePos (%d > %d), corruption, skipping to next writeFileNum and resetting 0...", d.name, d.readPos, d.writePos)
|
||
}
|
||
|
||
// 强制跳到下一个文件
|
||
d.skipToNextRWFile()
|
||
d.needSync = true
|
||
}
|
||
}
|
||
|
||
func (d *diskQueue) moveForward() {
|
||
oldReadFileNum := d.readFileNum
|
||
d.readFileNum = d.nextReadFileNum
|
||
d.readPos = d.nextReadPos
|
||
d.depth -= 1
|
||
|
||
//see if we need to clean up the old file
|
||
if oldReadFileNum != d.nextReadFileNum {
|
||
// sync every time we start reading from a new file
|
||
// 每次读取新文件时同步
|
||
d.needSync = true
|
||
|
||
fileName := d.fileName(oldReadFileNum)
|
||
if d.writeFile != nil && d.writeFile.Name() == fileName {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
}
|
||
err := os.Remove(fileName)
|
||
if err != nil {
|
||
d.logf(ERROR, "DISKQUEUE(%s) failed to Remove(%s) - %s", d.name, fileName, err)
|
||
}
|
||
}
|
||
d.checkTailCorruption(d.depth)
|
||
}
|
||
|
||
func (d *diskQueue) handleReadEOFError() {
|
||
fileName := d.fileName(d.readFileNum)
|
||
err := os.Remove(fileName)
|
||
if err != nil {
|
||
d.logf(ERROR, "DISKQUEUE(%s) failed to Remove(%s) - %s", d.name, fileName, err)
|
||
}
|
||
d.readFileNum++
|
||
d.readPos = 0
|
||
d.nextReadFileNum++
|
||
d.nextReadPos = 0
|
||
}
|
||
|
||
// 处理读取错误
|
||
func (d *diskQueue) handleReadError() {
|
||
// 跳转到下一个读取文件并重命名当前(损坏的)文件
|
||
if d.readFileNum == d.writeFileNum {
|
||
// 如果你不能正确地从当前的写文件中读取，那就应该跳过当前的文件
|
||
if d.writeFile != nil {
|
||
d.writeFile.Close()
|
||
d.writeFile = nil
|
||
}
|
||
d.writeFileNum++
|
||
d.writePos = 0
|
||
}
|
||
badFileName := d.fileName(d.readFileNum)
|
||
badRenameFilename := badFileName + ".bad"
|
||
d.logf(WARN, "DISKQUEUE(%s) jump to next file and saving bad file as %s", d.name, badRenameFilename)
|
||
err := os.Rename(badFileName, badRenameFilename)
|
||
if err != nil {
|
||
d.logf(ERROR, "DISKQUEUE(%s) failed to rename bad diskqueue file %s to %s", d.name, badFileName, badRenameFilename)
|
||
}
|
||
d.readFileNum++
|
||
d.readPos = 0
|
||
d.nextReadFileNum = d.readFileNum
|
||
d.nextReadPos = 0
|
||
|
||
// 重大状态变更，在下一次迭代中安排同步
|
||
d.needSync = true
|
||
|
||
d.checkTailCorruption(d.depth)
|
||
}
|
||
|
||
// ioLoop 提供了暴露go通道的后端(通过ReadChan())，以支持多个并发队列消费者
|
||
// 它的工作原理是基于队列是否有数据可读和阻塞，直到数据通过适当的go通道读取或写入
|
||
// 简单来讲，这也意味着我们正在异步地从文件系统读取数据
|
||
func (d *diskQueue) ioLoop() {
|
||
var (
|
||
dataRead []byte
|
||
err error
|
||
count int64
|
||
r chan []byte
|
||
p chan []byte
|
||
)
|
||
// 设置定时器
|
||
syncTicker := time.NewTicker(d.syncTimeout)
|
||
|
||
for {
|
||
// 若到达刷盘频次，标记等待刷盘
|
||
if count == d.syncEvery {
|
||
d.needSync = true
|
||
}
|
||
|
||
if d.needSync {
|
||
err = d.sync()
|
||
if err != nil {
|
||
d.logf(ERROR, "DISKQUEUE(%s) failed to sync - %s", d.name, err)
|
||
}
|
||
count = 0
|
||
}
|
||
// 有可读数据，并且当前读chan的数据已经被读走，则读取下一条数据
|
||
//if (d.readFileNum < d.writeFileNum) || (d.readPos < d.writePos) {
|
||
if d.readFileNum < d.writeFileNum || (d.readFileNum == d.writeFileNum && d.readPos < d.writePos) {
|
||
if d.nextReadPos == d.readPos {
|
||
dataRead, err = d.readOne()
|
||
if err != nil {
|
||
if io.EOF == err {
|
||
d.logf(WARN, "DISKQUEUE(%s) reading at %d of %s - %s", d.name, d.readPos, d.fileName(d.readFileNum), err)
|
||
// 读取了文件未尾，同时进行读写时可能会出现这个问题
|
||
d.handleReadEOFError()
|
||
} else {
|
||
d.logf(ERROR, "DISKQUEUE(%s) reading at %d of %s - %s", d.name, d.readPos, d.fileName(d.readFileNum), err)
|
||
d.handleReadError()
|
||
}
|
||
continue
|
||
}
|
||
}
|
||
r = d.readChan
|
||
p = d.peekChan
|
||
} else {
|
||
// 如果无可读数据，那么设置 r 为nil, 防止将dataRead数据重复传入readChan中
|
||
r = nil
|
||
p = nil
|
||
}
|
||
|
||
// Go通道规范规定，跳过select中的nil通道操作(读或写)，只有在有数据要读时，我们才将r设置为d.readChan
|
||
select {
|
||
// peekChan 不需要改变状态
|
||
case p <- dataRead:
|
||
// readChan 需要改变状态
|
||
case r <- dataRead:
|
||
count++
|
||
// 如果读readChan成功，则会修改读的偏移
|
||
d.moveForward()
|
||
case d.depthChan <- d.depth:
|
||
case <-d.emptyChan:
|
||
d.emptyResponseChan <- d.deleteAllFiles()
|
||
count = 0
|
||
case dataWrite := <-d.writeChan:
|
||
count++
|
||
d.writeResponseChan <- d.writeOne(dataWrite)
|
||
case <-syncTicker.C:
|
||
// 到刷盘时间，则修改needSync = true
|
||
if count == 0 {
|
||
// 没有活动时不要刷盘
|
||
continue
|
||
}
|
||
d.needSync = true
|
||
case <-d.exitChan:
|
||
// 退出
|
||
goto exit
|
||
}
|
||
}
|
||
|
||
exit:
|
||
d.logf(INFO, "DISKQUEUE(%s): closing ... ioLoop", d.name)
|
||
syncTicker.Stop()
|
||
d.exitSyncChan <- 1
|
||
}
|