nsq源碼分析

//原文：http://www.rzrgm.cn/hlxs/p/11445103.html 作者：啊漢

type Topic struct {
   // 64bit atomic vars need to be first for proper alignment on 32bit platforms
   messageCount uint64  //消息總數(shù)量
   messageBytes uint64  //消息總長(zhǎng)度
   sync.RWMutex
   name              string   //topic name
   channelMap        map[string]*Channel //保存topic下面的所有channel
   backend           BackendQueue //磁盤隊(duì)列
   memoryMsgChan     chan *Message //內(nèi)存隊(duì)列
   startChan         chan int
   exitChan          chan int
   channelUpdateChan chan int
   waitGroup         util.WaitGroupWrapper
   exitFlag          int32  //退出標(biāo)記
   idFactory         *guidFactory //生成msg id的工廠
   ephemeral      bool  //是否臨時(shí)topic
   deleteCallback func(*Topic)  //刪除topic方法指針
   deleter        sync.Once
   paused    int32   //暫停標(biāo)記，1暫停， 0正常
   pauseChan chan int
   ctx *context
}

// Command represents a command from a client to an NSQ daemon
//原文：http://www.rzrgm.cn/hlxs/p/11445103.html 作者：啊漢

type Command struct {
   Name   []byte   //命令名稱，可選：IDENTIFY、FIN、RDY、REQ、PUB、MPUB、DPUB、NOP、TOUCH、SUB、CLS、AUTH
   Params [][]byte //不同的命令做不同解析，涉及到topic的，Params[0]為topic name
   Body   []byte   //消息內(nèi)容
}
 
// WriteTo implements the WriterTo interface and
// serializes the Command to the supplied Writer.
//
// It is suggested that the target Writer is buffered
// to avoid performing many system calls.
func (c *Command) WriteTo(w io.Writer) (int64, error) {
   var total int64
   var buf [4]byte
 
   n, err := w.Write(c.Name)  //命名名稱，nsqd根據(jù)這個(gè)名稱執(zhí)行相關(guān)功能
   total += int64(n)
   if err != nil {
      return total, err
   }
 
   for _, param := range c.Params {
      n, err := w.Write(byteSpace)  //空格
      total += int64(n)
      if err != nil {
         return total, err
      }
      n, err = w.Write(param)  //參數(shù)
      total += int64(n)
      if err != nil {
         return total, err
      }
   }
 
   n, err = w.Write(byteNewLine)   //空行\(zhòng)n
   total += int64(n)
   if err != nil {
      return total, err
   }
 
   //消息內(nèi)容
   if c.Body != nil {   
      bufs := buf[:]
      binary.BigEndian.PutUint32(bufs, uint32(len(c.Body)))
      n, err := w.Write(bufs)   //消息長(zhǎng)度4字節(jié)
      total += int64(n)
      if err != nil {
         return total, err
      }
      n, err = w.Write(c.Body)  //消息內(nèi)容
      total += int64(n)
      if err != nil {
         return total, err
      }
   }
 
   return total, nil
}

func (p *protocolV2) Exec(client *clientV2, params [][]byte) ([]byte, error) {
   if bytes.Equal(params[0], []byte("IDENTIFY")) {
      return p.IDENTIFY(client, params)
   }
   err := enforceTLSPolicy(client, p, params[0])
   if err != nil {
      return nil, err
   }
   switch {
   case bytes.Equal(params[0], []byte("FIN")):
      return p.FIN(client, params)
   case bytes.Equal(params[0], []byte("RDY")):
      return p.RDY(client, params)
   case bytes.Equal(params[0], []byte("REQ")):
      return p.REQ(client, params)
   case bytes.Equal(params[0], []byte("PUB")):
      return p.PUB(client, params)
   case bytes.Equal(params[0], []byte("MPUB")):
      return p.MPUB(client, params)
   case bytes.Equal(params[0], []byte("DPUB")):
      return p.DPUB(client, params)
   case bytes.Equal(params[0], []byte("NOP")):
      return p.NOP(client, params)
   case bytes.Equal(params[0], []byte("TOUCH")):
      return p.TOUCH(client, params)
   case bytes.Equal(params[0], []byte("SUB")):
      return p.SUB(client, params)
   case bytes.Equal(params[0], []byte("CLS")):
      return p.CLS(client, params)
   case bytes.Equal(params[0], []byte("AUTH")):
      return p.AUTH(client, params)
   }
   return nil, protocol.NewFatalClientErr(nil, "E_INVALID", fmt.Sprintf("invalid command %s", params[0]))
}

 
type Channel struct {
   // 64bit atomic vars need to be first for proper alignment on 32bit platforms
   requeueCount uint64   //重新入隊(duì)數(shù)量
   messageCount uint64   //消息數(shù)量
   timeoutCount uint64   //超時(shí)數(shù)量，已經(jīng)消費(fèi)，但沒有反饋結(jié)果，會(huì)重新加入隊(duì)列，messageCount不會(huì)自增
 
   sync.RWMutex
 
   topicName string    //topic name
   name      string    //channel name
   ctx       *context
 
   backend BackendQueue  //將消息寫入磁盤的隊(duì)列，維護(hù)磁盤消息的讀寫
 
   memoryMsgChan chan *Message //內(nèi)存消息隊(duì)列，通道buffer默認(rèn)10000
   exitFlag      int32   //退出標(biāo)記，1表示退出，0沒有退出
   exitMutex     sync.RWMutex
 
   // state tracking
   clients        map[int64]Consumer   //連接到這個(gè)topic-channel的所有client
   paused         int32  //暫停標(biāo)記，0不暫停，1暫停，暫停就不會(huì)往這個(gè)channel中copy消息
   ephemeral      bool   //臨時(shí)channel標(biāo)記，臨時(shí)channel不會(huì)存到文件中
   deleteCallback func(*Channel) //用于從topic中刪除channel
   deleter        sync.Once  
 
   // Stats tracking
   e2eProcessingLatencyStream *quantile.Quantile
 
   // TODO: these can be DRYd up
   deferredMessages map[MessageID]*pqueue.Item  //延遲消息map，方便查找
   deferredPQ       pqueue.PriorityQueue  //延遲消息隊(duì)列
   deferredMutex    sync.Mutex
   inFlightMessages map[MessageID]*Message  //消費(fèi)中的消息map，方便查找
   inFlightPQ       inFlightPqueue   //消費(fèi)中的消息隊(duì)列
   inFlightMutex    sync.Mutex
}

// frame types
const (
   FrameTypeResponse int32 = 0   //響應(yīng)
   FrameTypeError    int32 = 1   //錯(cuò)誤
   FrameTypeMessage  int32 = 2   //消息
)

type inFlightPqueue []*Message  
inFlightPQ       inFlightPqueue //按照超時(shí)時(shí)間排序的最小堆
inFlightMessages map[MessageID]*Message //保存消息

type Item struct {
   Value    interface{}  //*Message
   Priority int64 //執(zhí)行的時(shí)間戳，單位毫秒
   Index    int   //隊(duì)列索引
}
type PriorityQueue []*Item
deferredPQ       pqueue.PriorityQueue
deferredMessages map[MessageID]*pqueue.Item
deferredPQ和inFlightPQ一樣，是按照時(shí)間排序的最小堆

type meta struct {
   Topics []struct {
      Name     string `json:"name"`
      Paused   bool   `json:"paused"`
      Channels []struct {
         Name   string `json:"name"`
         Paused bool   `json:"paused"`
      } `json:"channels"`
   } `json:"topics"`
}

// diskQueue implements a filesystem backed FIFO queue
//原文：http://www.rzrgm.cn/hlxs/p/11445103.html 作者：啊漢

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   //已經(jīng)讀的位置
   writePos     int64   //已經(jīng)寫的位置
   readFileNum  int64   //正在讀的文件編號(hào)
   writeFileNum int64   //正在寫的文件編號(hào)
   depth        int64   //沒有消費(fèi)的消息數(shù)量
 
   sync.RWMutex
 
   // instantiation time metadata
   name            string  // topicName 或者 topicName + ":" + channelName
   dataPath        string  //存消息文件的目錄
   maxBytesPerFile int64   // currently this cannot change once created
   minMsgSize      int32   //消息最小值
   maxMsgSize      int32   //消息最大值
   syncEvery       int64   // number of writes per fsync
   syncTimeout     time.Duration // duration of time per fsync
   exitFlag        int32   //退出標(biāo)記
   needSync        bool    //強(qiáng)制將文件緩沖區(qū)的數(shù)據(jù)寫入磁盤
 
   // keeps track of the position where we have read
   // (but not yet sent over readChan)
   nextReadPos     int64   //下次讀的位置
   nextReadFileNum int64   //下次讀的文件編號(hào)
 
   readFile  *os.File      //正在讀的文件
   writeFile *os.File      //正在寫的文件
   reader    *bufio.Reader //讀緩沖區(qū)，默認(rèn)4K
   writeBuf  bytes.Buffer  //寫緩沖區(qū)
 
   // exposed via ReadChan()
   readChan chan []byte    //讀channel
 
   // internal channels
   writeChan         chan []byte   //寫channel
   writeResponseChan chan error //寫結(jié)果通知
   emptyChan         chan int   //刪除所有文件channel
   emptyResponseChan chan error //刪除通知channel
   exitChan          chan int   //退出channel
   exitSyncChan      chan int   //退出命令同步等待channel
 
   logf AppLogFunc    //寫日志
}

func (d *diskQueue) fileName(fileNum int64) string {
   return fmt.Sprintf(path.Join(d.dataPath, "%s.diskqueue.%06d.dat"), d.name, fileNum)
}