//原文：https://www.cnblogs.com/hlxs/p/11445103.html 作者：啊汉
type Topic struct {// 64bit atomic vars need to be first for proper alignment on 32bit platformsmessageCount uint64  //消息总数量messageBytes uint64  //消息总长度sync.RWMutexname              string   //topic namechannelMap        map[string]*Channel //保存topic下面的所有channelbackend           BackendQueue //磁盘队列memoryMsgChan     chan *Message //内存队列startChan         chan intexitChan          chan intchannelUpdateChan chan intwaitGroup         util.WaitGroupWrapperexitFlag          int32  //退出标记idFactory         *guidFactory //生成msg id的工厂ephemeral      bool  //是否临时topicdeleteCallback func(*Topic)  //删除topic方法指针deleter        sync.Oncepaused    int32   //暂停标记，1暂停， 0正常pauseChan chan intctx *context
}

// Command represents a command from a client to an NSQ daemon//原文：https://www.cnblogs.com/hlxs/p/11445103.html 作者：啊汉
type Command struct {Name   []byte   //命令名称，可选：IDENTIFY、FIN、RDY、REQ、PUB、MPUB、DPUB、NOP、TOUCH、SUB、CLS、AUTHParams [][]byte //不同的命令做不同解析，涉及到topic的，Params[0]为topic nameBody   []byte   //消息内容
}// 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 int64var buf [4]byten, err := w.Write(c.Name)  //命名名称，nsqd根据这个名称执行相关功能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)  //参数total += int64(n)if err != nil {return total, err}}n, err = w.Write(byteNewLine)   //空行ntotal += int64(n)if err != nil {return total, err}//消息内容if c.Body != nil {   bufs := buf[:]binary.BigEndian.PutUint32(bufs, uint32(len(c.Body)))n, err := w.Write(bufs)   //消息长度4字节total += int64(n)if err != nil {return total, err}n, err = w.Write(c.Body)  //消息内容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 platformsrequeueCount uint64   //重新入队数量messageCount uint64   //消息数量timeoutCount uint64   //超时数量，已经消费，但没有反馈结果，会重新加入队列，messageCount不会自增sync.RWMutextopicName string    //topic namename      string    //channel namectx       *contextbackend BackendQueue  //将消息写入磁盘的队列，维护磁盘消息的读写memoryMsgChan chan *Message //内存消息队列，通道buffer默认10000exitFlag      int32   //退出标记，1表示退出，0没有退出exitMutex     sync.RWMutex// state trackingclients        map[int64]Consumer   //连接到这个topic-channel的所有clientpaused         int32  //暂停标记，0不暂停，1暂停，暂停就不会往这个channel中copy消息ephemeral      bool   //临时channel标记，临时channel不会存到文件中deleteCallback func(*Channel) //用于从topic中删除channeldeleter        sync.Once  // Stats trackinge2eProcessingLatencyStream *quantile.Quantile// TODO: these can be DRYd updeferredMessages map[MessageID]*pqueue.Item  //延迟消息map，方便查找deferredPQ       pqueue.PriorityQueue  //延迟消息队列deferredMutex    sync.MutexinFlightMessages map[MessageID]*Message  //消费中的消息map，方便查找inFlightPQ       inFlightPqueue   //消费中的消息队列inFlightMutex    sync.Mutex
}

// frame types
const (FrameTypeResponse int32 = 0   //响应FrameTypeError    int32 = 1   //错误FrameTypeMessage  int32 = 2   //消息
)

type inFlightPqueue []*Message
inFlightPQ       inFlightPqueue //按照超时时间排序的最小堆
inFlightMessages map[MessageID]*Message //保存消息

type Item struct {Value    interface{}  //*MessagePriority int64 //执行的时间戳，单位毫秒Index    int   //队列索引
}
type PriorityQueue []*Item
deferredPQ       pqueue.PriorityQueue
deferredMessages map[MessageID]*pqueue.Item
deferredPQ和inFlightPQ一样，是按照时间排序的最小堆

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//原文：https://www.cnblogs.com/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   //已经读的位置writePos     int64   //已经写的位置readFileNum  int64   //正在读的文件编号writeFileNum int64   //正在写的文件编号depth        int64   //没有消费的消息数量sync.RWMutex// instantiation time metadataname            string  // topicName 或者 topicName + ":" + channelNamedataPath        string  //存消息文件的目录maxBytesPerFile int64   // currently this cannot change once createdminMsgSize      int32   //消息最小值maxMsgSize      int32   //消息最大值syncEvery       int64   // number of writes per fsyncsyncTimeout     time.Duration // duration of time per fsyncexitFlag        int32   //退出标记needSync        bool    //强制将文件缓冲区的数据写入磁盘// keeps track of the position where we have read// (but not yet sent over readChan)nextReadPos     int64   //下次读的位置nextReadFileNum int64   //下次读的文件编号readFile  *os.File      //正在读的文件writeFile *os.File      //正在写的文件reader    *bufio.Reader //读缓冲区，默认4KwriteBuf  bytes.Buffer  //写缓冲区// exposed via ReadChan()readChan chan []byte    //读channel// internal channelswriteChan         chan []byte   //写channelwriteResponseChan chan error //写结果通知emptyChan         chan int   //删除所有文件channelemptyResponseChan chan error //删除通知channelexitChan          chan int   //退出channelexitSyncChan      chan int   //退出命令同步等待channellogf AppLogFunc    //写日志
}

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