vlambda博客
学习文章列表

type BackendQueue interface {
    Put([]byte) error // 将一条消息插入到队列中
    ReadChan() chan []byte // 返回一个无缓冲的chan
    Close() error // 队列关闭
    Delete() error // 删除队列 （实际在实现时，数据仍保留）
    Depth() int64          // 返回读延迟的消息量
    Empty() error // 清空消息 （实际会删除所有的记录文件）
}

// diskQueue implements a filesystem backed FIFO queue
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            // 每个文件最大字节数
    minMsgSize int32            // 单条消息的最小大小
    maxMsgSize int32            // 单挑消息的最大大小
    syncEvery int64            // 每写多少次刷盘一次
    syncTimeout time.Duration // 至少多久会刷盘一次
    exitFlag int32            // 退出标识
    needSync bool             // 如果 needSync 为true， 则需要fsync刷新metadata 数据

    // keeps track of the position where we have read
    // (but not yet sent over readChan)
    nextReadPos int64            // 下一次读的位置
    nextReadFileNum int64            // 下一次读的文件number


    readFile *os.File // 读 fd
    writeFile *os.File // 写 fd
    reader *bufio.Reader // 读 buffer
    writeBuf bytes.Buffer // 写 buffer

    // exposed via ReadChan()
    readChan chan []byte             // 读channel

    // internal channels
    writeChan chan []byte    // 写 channel
    writeResponseChan chan error // 同步写完之后的 response
    emptyChan chan int       // 清空文件的channel
    emptyResponseChan chan error // 同步清空文件后的channel
    exitChan chan int       // 退出channel
    exitSyncChan chan int       // 退出命令同步等待channel


    logf AppLogFunc // 日志句柄
}

func New(name string, dataPath string, maxBytesPerFile int64,
    minMsgSize int32, maxMsgSize int32,
    syncEvery int64, syncTimeout time.Duration, logf AppLogFunc) Interface {
    d := diskQueue{
        name: name,
        dataPath: dataPath,
        maxBytesPerFile: maxBytesPerFile,
        minMsgSize: minMsgSize,
        maxMsgSize: maxMsgSize,
        readChan: make(chan []byte),
        writeChan: make(chan []byte),
        writeResponseChan: make(chan error),
        emptyChan: make(chan int),
        emptyResponseChan: make(chan error),
        exitChan: make(chan int),
        exitSyncChan: make(chan int),
        syncEvery: syncEvery,
        syncTimeout: syncTimeout,
        logf: logf,
    }


    // no need to lock here, nothing else could possibly be touching this instance
    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
}

func (d *diskQueue) ioLoop() {
  var dataRead []byte
  var err error
  var count int64
  var r 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.nextReadPos == d.readPos {
        dataRead, err = d.readOne()
        if err != nil {
          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
    } else {
      // 如果无可读数据，那么设置 r 为nil, 防止将dataRead数据重复传入readChan中
      r = nil
    }

    select {
    // the Go channel spec dictates that nil channel operations (read or write)
    // in a select are skipped, we set r to d.readChan only when there is data to read
    case r <- dataRead:
      count++
      // moveForward sets needSync flag if a file is removed
      // 如果读chan 被写入成功，则会修改读的偏移
      d.moveForward()
    case <-d.emptyChan:
      // 清空所有文件，并返回empty的结果
      d.emptyResponseChan <- d.deleteAllFiles()
      count = 0
    case dataWrite := <-d.writeChan:
      // 写msg
      count++
      d.writeResponseChan <- d.writeOne(dataWrite)
    case <-syncTicker.C:
      // 到刷盘时间，则修改needSync = true
      if count == 0 {
        // avoid sync when there's no activity
        continue
      }
      d.needSync = true
    case <-d.exitChan:
      goto exit
    }
  }


exit:
  d.logf(INFO, "DISKQUEUE(%s): closing ... ioLoop", d.name)
  syncTicker.Stop()
  d.exitSyncChan <- 1
}

  
    
    
  

   
     
     
   
需要注意的点：
  
    
    
  

depth\nreadFileNum,readPos\nwriteFileNum,writePos

// 64bit atomic vars need to be first for proper alignment on 32bit platforms

// On ARM, x86-32, and 32-bit MIPS,
// it is the caller's responsibility to arrange for 64-bit
// alignment of 64-bit words accessed atomically. The first word in a
// variable or in an allocated struct, array, or slice can be relied upon to be
// 64-bit aligned.